python_configuration.html

<table class="configuration table table-bordered">
    <thead>
        <tr>
            <th class="text-left" style="width: 20%">Key</th>
            <th class="text-left" style="width: 15%">Default</th>
            <th class="text-left" style="width: 10%">Type</th>
            <th class="text-left" style="width: 55%">Description</th>
        </tr>
    </thead>
    <tbody>
        <tr>
            <td><h5>python.archives</h5></td>
            <td style="word-wrap: break-word;">(none)</td>
            <td>String</td>
            <td>Add python archive files for job. The archive files will be extracted to the working directory of python UDF worker. Currently only zip-format is supported. For each archive file, a target directory is specified. If the target directory name is specified, the archive file will be extracted to a name can directory with the specified name. Otherwise, the archive file will be extracted to a directory with the same name of the archive file. The files uploaded via this option are accessible via relative path. '#' could be used as the separator of the archive file path and the target directory name. Comma (',') could be used as the separator to specify multiple archive files. This option can be used to upload the virtual environment, the data files used in Python UDF. The data files could be accessed in Python UDF, e.g.: f = open('data/data.txt', 'r'). The option is equivalent to the command line option "-pyarch".</td>
        </tr>
        <tr>
            <td><h5>python.client.executable</h5></td>
            <td style="word-wrap: break-word;">"python"</td>
            <td>String</td>
            <td>The path of the Python interpreter used to launch the Python process when submitting the Python jobs via "flink run" or compiling the Java/Scala jobs containing Python UDFs. Equivalent to the environment variable PYFLINK_CLIENT_EXECUTABLE. The priority is as following: <br />1. the configuration 'python.client.executable' defined in the source code;<br />2. the environment variable PYFLINK_CLIENT_EXECUTABLE;<br />3. the configuration 'python.client.executable' defined in flink-conf.yaml</td>
        </tr>
        <tr>
            <td><h5>python.executable</h5></td>
            <td style="word-wrap: break-word;">"python"</td>
            <td>String</td>
            <td>Specify the path of the python interpreter used to execute the python UDF worker. The python UDF worker depends on Python 3.6+, Apache Beam (version == 2.27.0), Pip (version &gt;= 7.1.0) and SetupTools (version &gt;= 37.0.0). Please ensure that the specified environment meets the above requirements. The option is equivalent to the command line option "-pyexec".</td>
        </tr>
        <tr>
            <td><h5>python.files</h5></td>
            <td style="word-wrap: break-word;">(none)</td>
            <td>String</td>
            <td>Attach custom python files for job. The standard python resource file suffixes such as .py/.egg/.zip or directory are all supported. These files will be added to the PYTHONPATH of both the local client and the remote python UDF worker. Files suffixed with .zip will be extracted and added to PYTHONPATH. Comma (',') could be used as the separator to specify multiple files. The option is equivalent to the command line option "-pyfs". </td>
        </tr>
        <tr>
            <td><h5>python.fn-execution.arrow.batch.size</h5></td>
            <td style="word-wrap: break-word;">10000</td>
            <td>Integer</td>
            <td>The maximum number of elements to include in an arrow batch for Python user-defined function execution. The arrow batch size should not exceed the bundle size. Otherwise, the bundle size will be used as the arrow batch size.</td>
        </tr>
        <tr>
            <td><h5>python.fn-execution.bundle.size</h5></td>
            <td style="word-wrap: break-word;">100000</td>
            <td>Integer</td>
            <td>The maximum number of elements to include in a bundle for Python user-defined function execution. The elements are processed asynchronously. One bundle of elements are processed before processing the next bundle of elements. A larger value can improve the throughput, but at the cost of more memory usage and higher latency.</td>
        </tr>
        <tr>
            <td><h5>python.fn-execution.bundle.time</h5></td>
            <td style="word-wrap: break-word;">1000</td>
            <td>Long</td>
            <td>Sets the waiting timeout(in milliseconds) before processing a bundle for Python user-defined function execution. The timeout defines how long the elements of a bundle will be buffered before being processed. Lower timeouts lead to lower tail latencies, but may affect throughput.</td>
        </tr>
        <tr>
            <td><h5>python.fn-execution.memory.managed</h5></td>
            <td style="word-wrap: break-word;">true</td>
            <td>Boolean</td>
            <td>If set, the Python worker will configure itself to use the managed memory budget of the task slot. Otherwise, it will use the Off-Heap Memory of the task slot. In this case, users should set the Task Off-Heap Memory using the configuration key taskmanager.memory.task.off-heap.size.</td>
        </tr>
        <tr>
            <td><h5>python.map-state.iterate-response-batch-size</h5></td>
            <td style="word-wrap: break-word;">1000</td>
            <td>Integer</td>
            <td>The maximum number of the MapState keys/entries sent to Python UDF worker in each batch when iterating a Python MapState. Note that this is an experimental flag and might not be available in future releases.</td>
        </tr>
        <tr>
            <td><h5>python.map-state.read-cache-size</h5></td>
            <td style="word-wrap: break-word;">1000</td>
            <td>Integer</td>
            <td>The maximum number of cached entries for a single Python MapState. Note that this is an experimental flag and might not be available in future releases.</td>
        </tr>
        <tr>
            <td><h5>python.map-state.write-cache-size</h5></td>
            <td style="word-wrap: break-word;">1000</td>
            <td>Integer</td>
            <td>The maximum number of cached write requests for a single Python MapState. The write requests will be flushed to the state backend (managed in the Java operator) when the number of cached write requests exceed this limit. Note that this is an experimental flag and might not be available in future releases.</td>
        </tr>
        <tr>
            <td><h5>python.metric.enabled</h5></td>
            <td style="word-wrap: break-word;">true</td>
            <td>Boolean</td>
            <td>When it is false, metric for Python will be disabled. You can disable the metric to achieve better performance at some circumstance.</td>
        </tr>
        <tr>
            <td><h5>python.requirements</h5></td>
            <td style="word-wrap: break-word;">(none)</td>
            <td>String</td>
            <td>Specify a requirements.txt file which defines the third-party dependencies. These dependencies will be installed and added to the PYTHONPATH of the python UDF worker. A directory which contains the installation packages of these dependencies could be specified optionally. Use '#' as the separator if the optional parameter exists. The option is equivalent to the command line option "-pyreq".</td>
        </tr>
        <tr>
            <td><h5>python.state.cache-size</h5></td>
            <td style="word-wrap: break-word;">1000</td>
            <td>Integer</td>
            <td>The maximum number of states cached in a Python UDF worker. Note that this is an experimental flag and might not be available in future releases.</td>
        </tr>
    </tbody>
</table>