Realtime Compute for Apache Flink:Uso de dependencias de Python

Descripción general

Siga las instrucciones de las secciones a continuación para usar las dependencias de Python:

• Pre-installed Python environment

• Use a custom Python virtual environment

• Use a third-party Python package

• Use a JAR package

• Use built-in connectors, data formats, and catalogs

• Use data files

Entorno de Python preinstalado

El entorno completamente administrado de Flink incluye un entorno de Python preinstalado. Las versiones de Python son las siguientes:

• VVR 8.0.10 y anteriores: Python 3.7

• De VVR 8.0.11 hasta una versión anterior a VVR 11.7: Python 3.9

• VVR 11.7 y posteriores: Python 3.9, Python 3.10 y Python 3.11

Algunos paquetes de Python de terceros requieren versiones específicas de glibc. Las versiones de glibc preinstaladas en el entorno completamente administrado de Flink son las siguientes:

Uso de un entorno virtual de Python personalizado

Si el entorno de Python preinstalado no cumple con sus requisitos, utilice una versión personalizada de Python mediante entornos virtuales. Cada entorno virtual de Python proporciona un entorno de ejecución completo. Dentro de un entorno virtual, usted puede instalar varios paquetes de dependencia de Python. A continuación se describe cómo preparar un entorno virtual de Python.

1. Prepare un entorno virtual de Python.

   1. Prepare el script setup-pyflink-virtual-env.sh en su computadora local. El siguiente código muestra el contenido del script.

      X86

      set -e
      # Download the miniforge.sh script.
      wget "https://github.com/conda-forge/miniforge/releases/download/25.11.0-1/Miniforge3-25.11.0-1-Linux-x86_64.sh" -O "miniforge.sh"
      
      # Add execution permissions to the miniforge.sh script.
      chmod +x miniforge.sh
      
      # Install miniforge.
      ./miniforge.sh -b
      source /root/miniforge3/bin/activate
      
      # Create a Python virtual environment.
      mamba create -n venv python=3.10 -y
      eval "$(mamba shell hook --shell bash)"
      
      # Activate the Python virtual environment.
      mamba activate venv
      
      # Install the PyFlink dependency.
      pip install "ververica-flink==11.7.0"
      # For VVR 11.5 and earlier, no dedicated PyPI package is available. Install the open-source PyFlink instead:
      # pip install "apache-flink==1.20.3" "setuptools<81"
      
      # Remove unnecessary JAR files to reduce the package size.
      find /root/miniforge3/envs/venv/lib/python3.10/site-packages/pyflink/ -name *.jar | xargs rm
      
      # Deactivate the Conda Python virtual environment.
      mamba deactivate
      
      # Package the prepared Conda Python virtual environment.
      cd /root/miniforge3/envs/ && zip -r /root/venv.zip venv 

      ARM

      set -e
      # Download the miniforge.sh script.
      wget "https://github.com/conda-forge/miniforge/releases/download/25.11.0-1/Miniforge3-25.11.0-1-Linux-aarch64.sh" -O "miniforge.sh"
      
      # Add execution permissions to the miniforge.sh script.
      chmod +x miniforge.sh
      
      # Install miniforge.
      ./miniforge.sh -b
      source /root/miniforge3/bin/activate
      
      # Create a Python virtual environment.
      mamba create -n venv python=3.10 -y
      eval "$(mamba shell hook --shell bash)"
      
      # Activate the Python virtual environment.
      mamba activate venv
      
      # Install the PyFlink dependency.
      yum install -y java-11-openjdk-devel
      export JAVA_HOME=/usr/lib/jvm/java-11
      wget "https://raw.githubusercontent.com/apache/flink/release-1.20/flink-python/dev/dev-requirements.txt" -O dev-requirements.txt
      pip install -r dev-requirements.txt
      pip install "ververica-flink==11.7.0"
      # For VVR 11.5 and earlier, no dedicated PyPI package is available. Install the open-source PyFlink instead:
      # pip install "apache-flink==1.20.3" "setuptools<81"
      
      # Remove unnecessary JAR files to reduce the package size.
      find /root/miniforge3/envs/venv/lib/python3.10/site-packages/pyflink/ -name *.jar | xargs rm
      
      # Deactivate the Conda Python virtual environment.
      mamba deactivate
      
      # Package the prepared Conda Python virtual environment.
      cd /root/miniforge3/envs && zip -r /root/venv.zip venv

      Nota

      En este tema, el despliegue utiliza VVR 11.7 y se ejecuta en un entorno virtual de Python 3.10. Si desea usar otra versión de VVR o instalar un entorno virtual con una versión distinta de Python, modifique los siguientes parámetros:

      • mamba create: cambie este valor por la versión de Python que necesite.

      • pip install:

        • VVR 11.6 y posteriores: instale ververica-flink y ajuste la versión para que coincida con la versión de VVR de su despliegue.

        • VVR 11.5 y anteriores: instale apache-flink y ajuste la versión a la de Flink correspondiente a la versión de VVR de su despliegue. Para obtener más información sobre cómo consultar la versión de Flink, consulte Storage management.

   2. Prepare el script build.sh en su computadora local. El siguiente código muestra el contenido del script.

      #!/bin/bash
      set -e -x
      
      yum install -y zip wget
      
      cd /root/
      bash /build/setup-pyflink-virtual-env.sh
      mv venv.zip /build/

   3. En la CLI, ejecute el siguiente comando para instalar el entorno virtual de Python:

      X86

      docker run -it --rm -v $PWD:/build -w /build quay.io/pypa/manylinux_2_28_x86_64 bash ./build.sh

      ARM

      docker run -it --rm -v $PWD:/build -w /build quay.io/pypa/manylinux_2_28_aarch64 bash ./build.sh

      Después de ejecutar el comando, se genera un archivo llamado venv.zip. En este ejemplo se utiliza el entorno virtual de Python 3.10.

      También puede modificar el script anterior para instalar el paquete de Python de terceros que necesite dentro del entorno virtual.

2. Utilice el entorno virtual de Python en los despliegues de Python.

   1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

   2. En la pestaña Fully Managed Flink, busque el espacio de trabajo que desea gestionar y haga clic en Console en la columna Actions.

   3. En el panel de navegación izquierdo, haga clic en Artifacts. En la página Artifacts, haga clic en Upload Artifact. En el cuadro de diálogo que aparece, seleccione el paquete venv.zip.

   4. En la página O&M > Deployments, haga clic en el nombre del trabajo deseado.

   5. En la pestaña Configuration, haga clic en Edit en la esquina superior derecha de la sección Basic y seleccione el paquete venv.zip en la lista desplegable Python Archives.

      Si el despliegue es de tipo SQL y necesita usar funciones definidas por el usuario (UDF) de Python, haga clic en Edit en la esquina superior derecha de la sección Parameters y agregue la siguiente configuración al campo Other Configuration:

      python.archives: oss://.../venv.zip

   6. En la sección Parameters, agregue la información de configuración de la ruta donde se instalará el entorno virtual de Python especificado al campo Other Configuration. Esta ruta depende de la versión de VVR de su despliegue.

      • VVR 6.X o posterior

        python.executable: venv.zip/venv/bin/python
        python.client.executable: venv.zip/venv/bin/python

      • Una versión del motor anterior a VVR 6.X

        python.executable: venv.zip/venv/bin/python

Uso de un paquete de Python de terceros

Uso de un paquete JAR

Si su despliegue de Python utiliza clases de Java, como un conector o una UDF de Java, siga estos pasos para especificar el paquete JAR del conector o de la UDF de Java.

1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

2. En la pestaña Fully Managed Flink, busque el espacio de trabajo que desea gestionar y haga clic en Console en la columna Actions.

3. En el panel de navegación izquierdo, haga clic en Artifacts. En la página Artifacts, haga clic en Upload Artifact. En el cuadro de diálogo, seleccione el paquete JAR que desea usar.

4. En la página Deployments, haga clic en el nombre del despliegue deseado. En la pestaña Configuration, haga clic en Edit en la esquina superior derecha de la sección Basic y seleccione el paquete JAR necesario en la lista desplegable Additional Dependencies.

5. En la pestaña Configuration, haga clic en Edit en la esquina superior derecha de la sección Parameters y agregue la siguiente configuración al campo Other Configuration.

   Por ejemplo, si el borrador depende de dos paquetes JAR llamados jar1.jar y jar2.jar, agregue la siguiente información de configuración:

   pipeline.classpaths: 'file:///flink/usrlib/jar1.jar;file:///flink/usrlib/jar2.jar'

6. Haga clic en Save.

Uso de conectores, formatos de datos y catálogos integrados

Para usar conectores, formatos de datos y catálogos integrados en un trabajo de Python de Flink, especifíquelos de la siguiente manera.

1. En la sección Parameters, agregue la configuración al campo Other Configuration.

   Para depender de varios conectores integrados llamados kafka y sls, agregue la siguiente configuración. Para consultar los nombres de los conectores integrados, revise la documentación de cada conector en Supported connectors.

   pipeline.used-builtin-connectors: kafka;sls

   Para depender de varios formatos de datos integrados llamados avro y parquet, agregue la siguiente configuración. Para consultar los nombres de los formatos de datos integrados, revise la documentación de los Data formats compatibles.

   pipeline.used-builtin-formats: avro;parquet

   Para depender de varios catálogos integrados llamados hive-2.3.6 y paimon, agregue la siguiente configuración. Para consultar los nombres de los catálogos integrados, revise la documentación de cada catálogo en Catalogs.

   pipeline.used-builtin-catalogs: hive-2.3.6;paimon

2. Haga clic en Save.

Uso de archivos de datos

Referencias

• Para obtener más información sobre cómo desarrollar un borrador de API de Python, consulte Develop PyFlink jobs.

• Para obtener más información sobre cómo desarrollar un despliegue de Python de Realtime Compute for Apache Flink, consulte PyFlink job.

• Fully managed Flink es compatible con borradores SQL y borradores DataStream. Para obtener más información sobre cómo desarrollarlos, consulte Job development overview y Develop a JAR job.