Componente opcional de Flink de Dataproc

Puedes activar componentes adicionales, como Flink, cuando creas un clúster de Dataproc con la función de componentes opcionales. En esta página, se muestra cómo crear un clúster de Dataproc con el componente opcional Apache Flink activado (un clúster de Flink) y, luego, ejecutar trabajos de Flink en el clúster.

Puedes usar tu clúster de Flink para lo siguiente:

1. Ejecuta trabajos de Flink con el recurso Jobs de Dataproc desde la consola, Google Cloud CLI o la API de Dataproc. Google Cloud

2. Ejecuta trabajos de Flink con la CLI de flink en el nodo instancia principal del clúster de Flink.

3. Ejecuta trabajos de Apache Beam en Flink

4. Ejecuta Flink en un clúster de Kerberized

Crea un clúster de Dataproc Flink

Puedes usar la consola de Google Cloud , Google Cloud CLI o la API de Dataproc para crear un clúster de Dataproc que tenga el componente de Flink activado.

Recomendación: Usa un clúster de VM estándar de 1 principal con el componente de Flink. Los clústeres en modo de alta disponibilidad de Dataproc (con 3 VMs principales) no admiten el modo de alta disponibilidad de Flink.

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --image-version=DATAPROC_IMAGE_VERSION \
    --optional-components=FLINK \
    --enable-component-gateway \
    --properties=PROPERTIES
    ... other flags

Después de crear un clúster de Flink

• Usa el vínculo Flink History Server en la puerta de enlace de componentes para ver el servidor de historial de Flink que se ejecuta en el clúster de Flink.
• Usa YARN ResourceManager link en la puerta de enlace de componentes para ver la interfaz web del administrador de trabajos de Flink que se ejecuta en el clúster de Flink .
• Crea un servidor de historial persistente de Dataproc para ver los archivos del historial de trabajos de Flink escritos por clústeres de Flink existentes y borrados.

Ejecuta trabajos de Flink con el recurso Jobs de Dataproc

Puedes ejecutar trabajos de Flink con el recurso Jobs de Dataproc desde laGoogle Cloud consola, Google Cloud CLI o la API de Dataproc.

gcloud dataproc jobs submit flink \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --class=MAIN_CLASS \
    --jar=JAR_FILE \
    -- JOB_ARGS

POST https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit

{
  "job": {
    "placement": {
      "clusterName": "CLUSTER_NAME"
    },
    "flinkJob": {
      "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
      "jarFileUris": [
        "file:///usr/lib/flink/examples/batch/WordCount.jar"
      ]
    }
  }
}

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit"

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit" | Select-Object -Expand Content

{
  "reference": {
    "projectId": "PROJECT_ID",
    "jobId": "JOB_ID"
  },
  "placement": {
    "clusterName": "CLUSTER_NAME",
    "clusterUuid": "CLUSTER_UUID"
  },
  "flinkJob": {
    "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
    "args": [
      "1000"
    ],
    "jarFileUris": [
      "file:///usr/lib/flink/examples/batch/WordCount.jar"
    ]
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "2020-10-07T20:16:21.759Z"
  },
  "jobUuid": "JOB_UUID"
}

Ejecuta trabajos de Flink con la CLI de flink

En lugar de ejecutar trabajos de Flink con el recurso Jobs de Dataproc, puedes ejecutar trabajos de Flink en el nodo principal de tu clúster de Flink con la CLI de flink.

En las siguientes secciones, se describen diferentes formas de ejecutar un trabajo de la CLI de flink en tu clúster de Dataproc Flink.

1. Establece una conexión SSH con el nodo principal: Usa la utilidad SSH para abrir una ventana de terminal en la VM instancia principal del clúster.

2. Establece la ruta de acceso de clase: Inicializa la ruta de acceso de clase de Hadoop desde la ventana de la terminal SSH en la VM instancia principal del clúster de Flink:

   export HADOOP_CLASSPATH=$(hadoop classpath)
   

3. Ejecuta trabajos de Flink: Puedes ejecutar trabajos de Flink en diferentes modos de implementación en YARN: modo de aplicación, por trabajo y de sesión.

   1. Modo de aplicación: El modo de aplicación de Flink es compatible con la versión 2.0 y posteriores de la imagen de Dataproc. Este modo ejecuta el método main() del trabajo en el administrador de trabajos de YARN. El clúster se apaga después de que finaliza el trabajo.

      Ejemplo de envío de trabajo:

      flink run-application \
          -t yarn-application \
          -Djobmanager.memory.process.size=1024m \
          -Dtaskmanager.memory.process.size=2048m \
          -Djobmanager.heap.mb=820 \
          -Dtaskmanager.heap.mb=1640 \
          -Dtaskmanager.numberOfTaskSlots=2 \
          -Dparallelism.default=4 \
          /usr/lib/flink/examples/batch/WordCount.jar
      

      Enumera los trabajos en ejecución:

      ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
      

      Para cancelar un trabajo en ejecución, haz lo siguiente:

      ./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
      

   2. Modo por trabajo: Este modo de Flink ejecuta el método main() del trabajo en el cliente.

      Ejemplo de envío de trabajo:

      flink run \
          -m yarn-cluster \
          -p 4 \
          -ys 2 \
          -yjm 1024m \
          -ytm 2048m \
          /usr/lib/flink/examples/batch/WordCount.jar
      

   3. Modo de sesión: Inicia una sesión de Flink en YARN de larga duración y, luego, envía uno o más trabajos a la sesión.

      1. Inicia una sesión: Puedes iniciar una sesión de Flink de una de las siguientes maneras:

         1. Crea un clúster de Flink y agrega la marca --metadata flink-start-yarn-session=true al comando gcloud dataproc clusters create (consulta Crea un clúster de Dataproc Flink). Con esta marca habilitada, después de que se crea el clúster, Dataproc ejecuta /usr/bin/flink-yarn-daemon para iniciar una sesión de Flink en el clúster.

            El ID de la aplicación de YARN de la sesión se guarda en /tmp/.yarn-properties-${USER}. Puedes solicitar el ID con el comando yarn application -list.

         2. Ejecuta la secuencia de comandos de Flink yarn-session.sh, que está preinstalada en la VM instancia principal del clúster, con parámetros de configuración personalizados:

            Ejemplo con parámetros de configuración personalizados:

            /usr/lib/flink/bin/yarn-session.sh \
                -s 1 \
                -jm 1024m \
                -tm 2048m \
                -nm flink-dataproc \
                --detached
            

         3. Ejecuta la secuencia de comandos de wrapper de /usr/bin/flink-yarn-daemon de Flink con la configuración predeterminada:

            . /usr/bin/flink-yarn-daemon
            

      2. Envía un trabajo a una sesión: Ejecuta el siguiente comando para enviar un trabajo de Flink a la sesión.

         flink run -m <var>FLINK_MASTER_URL</var>/usr/lib/flink/examples/batch/WordCount.jar
         

         • FLINK_MASTER_URL: Es la URL, incluidos el host y el puerto, de la VM principal de Flink en la que se ejecutan los trabajos. Quita el http:// prefix de la URL. Esta URL se muestra en el resultado del comando cuando inicias una sesión de Flink. Puedes ejecutar el siguiente comando para enumerar esta URL en el campo Tracking-URL:

         yarn application -list -appId=<yarn-app-id> | sed 's#http://##'
            ```
         

      3. Enumera los trabajos en una sesión: Para enumerar los trabajos de Flink en una sesión, haz una de las siguientes acciones:

         • Ejecuta flink list sin argumentos. El comando busca el ID de la aplicación de YARN de la sesión en /tmp/.yarn-properties-${USER}.

         • Obtén el ID de la aplicación de YARN de la sesión desde /tmp/.yarn-properties-${USER} o el resultado de yarn application -list y, luego, ejecuta <code>flink list -yid YARN_APPLICATION_ID.

         • Ejecuta flink list -m FLINK_MASTER_URL.

      4. Detén una sesión: Para detener la sesión, obtén el ID de la aplicación de YARN de la sesión desde /tmp/.yarn-properties-${USER} o el resultado de yarn application -list y, luego, ejecuta cualquiera de los siguientes comandos:

         echo "stop" | /usr/lib/flink/bin/yarn-session.sh -id YARN_APPLICATION_ID
         

         yarn application -kill YARN_APPLICATION_ID
         

Ejecuta trabajos de Apache Beam en Flink

Puedes ejecutar trabajos de Apache Beam en Dataproc mediante FlinkRunner.

Puedes ejecutar trabajos de Beam en Flink de las siguientes maneras:

1. Trabajos de Java Beam
2. Trabajos de Portable Beam

Trabajos de Java Beam

Empaqueta tus trabajos de Beam en un archivo JAR. Proporciona el archivo JAR empaquetado con las dependencias necesarias para ejecutar el trabajo.

En el siguiente ejemplo, se ejecuta un trabajo de Java Beam desde el nodo principal del clúster de Dataproc.

1. Crea un clúster de Dataproc con el componente Flink habilitado.

   gcloud dataproc clusters create CLUSTER_NAME \
       --optional-components=FLINK \
       --image-version=DATAPROC_IMAGE_VERSION \
       --region=REGION \
       --enable-component-gateway \
       --scopes=https://www.googleapis.com/auth/cloud-platform
   

   • --optional-components: Flink.
   • --image-version: Es la versión de la imagen del clúster, que determina la versión de Flink instalada en el clúster (por ejemplo, consulta las versiones de componente de Apache Flink enumeradas para las cuatro versiones 2.0.x más recientes y la cuatro anteriores).
   • --region: Es una región de Dataproc compatible.
   • --enable-component-gateway: Habilita el acceso a la IU del administrador de trabajo de Flink.
   • --scopes: Habilita el acceso a las Google Cloud APIs de tu clúster (consulta Prácticas recomendadas sobre permisos). El alcance de cloud-platform está habilitado de forma predeterminada (no es necesario que incluyas este parámetro de configuración de la marca) cuando creas un clúster que usa la versión de imagen 2.1 de Dataproc o una posterior.

2. Usa la utilidad de SSH para abrir una ventana de terminal en el nodo instancia principal del clúster de Flink.

3. Inicia una sesión de Flink en YARN en el nodo instancia principal del clúster de Dataproc.

   . /usr/bin/flink-yarn-daemon
   

   Toma nota de la versión de Flink en tu clúster de Dataproc.

   flink --version
   

4. En tu máquina local, genera el ejemplo de recuento de palabras canónicos de Beam en Java.

   Elige una versión de Beam compatible con la versión de Flink en tu clúster de Dataproc. Consulta la tabla Compatibilidad de versiones de Flink que enumera la compatibilidad de versiones de Beam-Flink.

   Abre el archivo POM generado. Verifica la versión del ejecutor de Flink de Beam que especifica la etiqueta <flink.artifact.name>. Si la versión del ejecutor de Flink de Beam en el nombre del artefacto de Flink no coincide con la versión de Flink en tu clúster, actualiza el número de versión para que coincida.

   mvn archetype:generate \
       -DarchetypeGroupId=org.apache.beam \
       -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
       -DarchetypeVersion=BEAM_VERSION \
       -DgroupId=org.example \
       -DartifactId=word-count-beam \
       -Dversion="0.1" \
       -Dpackage=org.apache.beam.examples \
       -DinteractiveMode=false
   

5. Empaqueta el ejemplo de conteo de palabras.

   mvn package -Pflink-runner
   

6. Sube el archivo uber JAR empaquetado, word-count-beam-bundled-0.1.jar (~135 MB) al nodo principal del clúster de Dataproc. Puedes usar gcloud storage cp para transferencias de archivos más rápidas a tu clúster de Dataproc desde Cloud Storage.

   1. En tu terminal local, crea un bucket de Cloud Storage y sube el archivo uber JAR.

      gcloud storage buckets create BUCKET_NAME
      
      gcloud storage cp target/word-count-beam-bundled-0.1.jar gs://BUCKET_NAME/
      

   2. En el nodo principal de Dataproc, descarga el archivo uber JAR.

      gcloud storage cp gs://BUCKET_NAME/word-count-beam-bundled-0.1.jar .
      

7. Ejecutar el trabajo de Java Beam en el nodo principal del clúster de Dataproc

   flink run -c org.apache.beam.examples.WordCount word-count-beam-bundled-0.1.jar \
       --runner=FlinkRunner \
       --output=gs://BUCKET_NAME/java-wordcount-out
   

8. Comprueba que los resultados se hayan escrito en tu bucket de Cloud Storage.

   gcloud storage cat gs://BUCKET_NAME/java-wordcount-out-SHARD_ID
   

9. Detén la sesión de Flink en YARN.

   yarn application -list
   
   yarn application -kill YARN_APPLICATION_ID
   

Trabajos de Portable Beam

Para ejecutar trabajos de Beam escritos en Python, Go y otros lenguajes compatibles, puedes usar FlinkRunner y PortableRunner como se describe en la página del ejecutor de Flink de Beam (también consulta la hoja de ruta del marco de trabajo de portabilidad).

En el siguiente ejemplo, se ejecuta un trabajo portátil de Beam en Python desde el nodo principal del clúster de Dataproc.

1. Crea un clúster de Dataproc con los componentes Flink y Docker habilitados.

   gcloud dataproc clusters create CLUSTER_NAME \
       --optional-components=FLINK,DOCKER \
       --image-version=DATAPROC_IMAGE_VERSION \
       --region=REGION \
       --enable-component-gateway \
       --scopes=https://www.googleapis.com/auth/cloud-platform
   

   Notas:

   • --optional-components: Flink y Docker.
   • --image-version: Es la versión de la imagen del clúster, que determina la versión de Flink instalada en el clúster (por ejemplo, consulta las versiones de componente de Apache Flink enumeradas para las cuatro versiones 2.0.x más recientes y las cuatro anteriores).
   • --region: Es una región de Dataproc disponible.
   • --enable-component-gateway: Habilita el acceso a la IU del administrador de trabajo de Flink.
   • --scopes: Habilita el acceso a las APIs de tu clúster (consulta la práctica recomendada sobre permisos). Google Cloud El alcance de cloud-platform está habilitado de forma predeterminada (no es necesario que incluyas este parámetro de configuración de la marca) cuando creas un clúster que usa la versión de imagen 2.1 de Dataproc o una posterior.

2. Usa gcloud CLI de forma local o en Cloud Shell para crear un bucket de Cloud Storage. Especificarás BUCKET_NAME cuando ejecutes un programa de recuento de palabras de ejemplo.

   gcloud storage buckets create BUCKET_NAME
   

3. En una ventana de terminal de la VM del clúster, inicia una sesión de Flink en YARN. Anota la URL de la instancia principal de Flink, la dirección de la instancia principal de Flink en la que se ejecutan los trabajos. Especificarás FLINK_MASTER_URL cuando ejecutes un programa de recuento de palabras de muestra.

   . /usr/bin/flink-yarn-daemon
   

   Muestra y toma nota de la versión de Flink que ejecuta el clúster de Dataproc. Especificarás FLINK_VERSION cuando ejecutes un programa de recuento de palabras de muestra.

   flink --version
   

4. Instala las bibliotecas de Python necesarias para el trabajo en el nodo instancia principal del clúster.

5. Instala una versión de Beam que sea compatible con la versión de Flink en el clúster.

   python -m pip install apache-beam[gcp]==BEAM_VERSION
   

6. Ejecuta el ejemplo de recuento de palabras en el nodo instancia principal del clúster.

   python -m apache_beam.examples.wordcount \
       --runner=FlinkRunner \
       --flink_version=FLINK_VERSION \
       --flink_master=FLINK_MASTER_URL
       --flink_submit_uber_jar \
       --output=gs://BUCKET_NAME/python-wordcount-out
   

   Notas:

   • --runner: FlinkRunner.
   • --flink_version: FLINK_VERSION, que se mencionó anteriormente.
   • --flink_master: FLINK_MASTER_URL, que se mencionó anteriormente.
   • --flink_submit_uber_jar: Usa el uber JAR para ejecutar el trabajo de Beam.
   • --output: BUCKET_NAME, que creaste antes.

7. Verifica que los resultados se hayan escrito en tu bucket.

   gcloud storage cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID
   

8. Detén la sesión de Flink en YARN.

   1. Obtén el ID de la aplicación.

   yarn application -list
   
   1. Insert the <var>YARN_APPLICATION_ID</var>, then stop the session.
   
   yarn application -kill 
   

Ejecuta Flink en un clúster de Kerberized

El componente Flink de Dataproc admite clústeres con Kerberos. Se necesita un ticket Kerberos válido para enviar y continuar un trabajo de Flink o iniciar un clúster de Flink. De forma predeterminada, un ticket de Kerberos sigue siendo válido durante siete días.

Accede a la IU del administrador de trabajos de Flink

La interfaz web del administrador de trabajo de Flink está disponible mientras se ejecuta un clúster de Flink o un clúster de sesión de Flink. Para usar la interfaz web, haz lo siguiente:

1. Crea un clúster de Dataproc Flink.
2. Después de la creación del clúster, haz clic en el vínculo ResourceManager de YARN de la puerta de enlace de componentes en la pestaña de interfaz web en la página Detalles del clúster en la consola de Google Cloud .
3. En la IU de Resource Manager de YARN, identifica la entrada de la aplicación del clúster de Flink. Según el estado de finalización de un trabajo, se mostrará el vínculo ApplicationMaster o History.

   

4. En un trabajo de transmisión de larga duración, haz clic en el vínculo ApplicationManager para abrir el panel de Flink; para un trabajo completado, haz clic en el vínculo Historial para ver los detalles del trabajo.