Puedes activar componentes adicionales, como Flink, cuando creas un clúster de Dataproc con la función 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 hacer lo siguiente:
Ejecuta trabajos de Flink con el recurso
Jobs
de Dataproc desde la consola de Google Cloud, Google Cloud CLI o la API de Dataproc.Ejecuta trabajos de Flink con la CLI de
flink
que se ejecuta en el nodo de la instancia principal del clúster de Flink.Ejecuta Flink en un clúster kerberizado
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 Flink activado en el clúster.
Recomendación: Usa un clúster de VM principal de 1 instancia estándar 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.
Console
Para crear un clúster de Flink de Dataproc con la consola de Google Cloud, sigue estos pasos:
Abre la página Crea un clúster de Dataproc en Compute Engine de Dataproc.
- Se selecciona el panel Configurar clúster.
- En la sección Control de versiones, confirma o cambia el Tipo de imagen y versión. La versión de la imagen del clúster determina la versión del componente de Flink instalado en el clúster.
- La versión de la imagen debe ser 1.5 o superior para activar el componente de Flink en el clúster (consulta las versiones compatibles de Dataproc para ver las listas de las versiones de los componentes incluidas en cada versión de la imagen de Dataproc).
- La versión de la imagen debe ser [TBD] o superior para ejecutar trabajos de Flink a través de la API de trabajos de Dataproc (consulta Ejecuta trabajos de Flink de Dataproc).
- En la sección Componentes, haz lo siguiente:
- En Puerta de enlace del componente, selecciona Habilitar puerta de enlace de componentes. Debes habilitar la puerta de enlace de componentes para activar el vínculo de la puerta de enlace de componentes a la IU del servidor de historial de Flink. Habilitar la puerta de enlace de componentes también habilita el acceso a la interfaz web del administrador de trabajos de Flink que se ejecuta en el clúster de Flink.
- En Componentes opcionales, selecciona Flink y otros componentes opcionales para activar en el clúster.
- En la sección Control de versiones, confirma o cambia el Tipo de imagen y versión. La versión de la imagen del clúster determina la versión del componente de Flink instalado en el clúster.
Haz clic en el panel Personalizar clúster (opcional).
En la sección Propiedades del clúster, haz clic en Agregar propiedades para cada propiedad de clúster opcional que desees agregar a tu clúster. Puedes agregar propiedades con el prefijo
flink
para configurar las propiedades de Flink en/etc/flink/conf/flink-conf.yaml
, que actuarán como valores predeterminados para las aplicaciones de Flink que ejecutes en el clúster.Ejemplos:
- Configura
flink:historyserver.archive.fs.dir
para especificar la ubicación de Cloud Storage en la que se escriben los archivos de historial de trabajos de Flink (el servidor de historial de Flink que se ejecuta en el clúster de Flink usará esta ubicación). - Configura espacios para tareas de Flink con
flink:taskmanager.numberOfTaskSlots=n
.
- Configura
En la sección Metadatos de clúster personalizados, haz clic en Agregar metadatos para agregar metadatos opcionales. Por ejemplo, agrega
flink-start-yarn-session
true
para ejecutar el daemon de YARN de Flink (/usr/bin/flink-yarn-daemon
) en segundo plano, en el nodo instancia principal del clúster para iniciar una sesión de YARN de Flink (consulta Modo de sesión de Flink).
Si usas la versión 2.0 o anterior de la imagen de Dataproc, haz clic en el panel Administrar la seguridad (opcional) y, luego, en Acceso al proyecto, selecciona
Enables the cloud-platform scope for this cluster
. El permisocloud-platform
se habilita de forma predeterminada cuando creas un clúster que usa la versión 2.1 o posterior con la imagen de Dataproc.
- Se selecciona el panel Configurar clúster.
Haz clic en Crear para generar el clúster.
gcloud
Para crear un clúster de Dataproc Flink con gcloud CLI, ejecuta el siguiente comando gcloud dataproc clusters create de forma local en una ventana de terminal o en Cloud Shell:
gcloud dataproc clusters create CLUSTER_NAME \ --region=REGION \ --image-version=DATAPROC_IMAGE_VERSION \ --optional-components=FLINK \ --enable-component-gateway \ --properties=PROPERTIES ... other flags
Notas:
- CLUSTER_NAME: Especifica el nombre del clúster.
- REGION: Especifica una región de Compute Engine en la que se ubicará el clúster.
DATAPROC_IMAGE_VERSION: De manera opcional, especifica la versión de la imagen que se usará en el clúster. La versión de la imagen del clúster determina la versión del componente de Flink instalado en el clúster.
La versión de la imagen debe ser 1.5 o superior para activar el componente de Flink en el clúster (consulta las versiones compatibles de Dataproc para ver las listas de las versiones de los componentes incluidas en cada versión de la imagen de Dataproc).
La versión de la imagen debe ser [TBD] o superior para ejecutar trabajos de Flink a través de la API de trabajos de Dataproc (consulta Ejecuta trabajos de Flink de Dataproc).
--optional-components
: Debes especificar el componenteFLINK
para ejecutar los trabajos de Flink y el servicio web de HistoryServer de Flink en el clúster.--enable-component-gateway
: Debes habilitar la puerta de enlace de componentes para activar el vínculo de la puerta de enlace de componentes a la IU del servidor de historial de Flink. Habilitar la puerta de enlace de componentes también habilita el acceso a la interfaz web del administrador de trabajos de Fusion que se ejecuta en el clúster de Fusion.PROPERTIES. De manera opcional, especifica una o más propiedades del clúster.
Cuando creas clústeres de Dataproc con versiones de imagen
2.0.67
y2.1.15
o superiores, puedes usar la marca--properties
para configurar las propiedades de Flink en/etc/flink/conf/flink-conf.yaml
, que actuarán como predeterminadas para las aplicaciones de Flink que ejecutes en el clúster.Puedes configurar
flink:historyserver.archive.fs.dir
para especificar la ubicación de Cloud Storage en la que se escriben los archivos de historial de trabajos de Flink (el servidor de historial de Flink que se ejecuta en el clúster de Flink usará esta ubicación).Ejemplo de varias propiedades:
--properties=flink:historyserver.archive.fs.dir=gs://my-bucket/my-flink-cluster/completed-jobs,flink:taskmanager.numberOfTaskSlots=2
Otras marcas:
- Puedes agregar la marca
--metadata flink-start-yarn-session=true
opcional para ejecutar el daemon de YARN de Flink (/usr/bin/flink-yarn-daemon
) en segundo plano, en el nodo instancia principal del clúster para iniciar una sesión de YARN de Flink (consulta Modo de sesión de Flink).
- Puedes agregar la marca
Cuando usas la versión 2.0 o versiones anteriores, puedes agregar la marca
--scopes=https://www.googleapis.com/auth/cloud-platform
para habilitar el acceso de tu clúster a las API de Google Cloud (consulta las Prácticas recomendadas de permisos). El permisocloud-platform
se habilita de forma predeterminada cuando creas un clúster que usa la versión 2.1 o posterior con la imagen de Dataproc.
API
Para crear un clúster de Flink de Dataproc con la API de Dataproc, envía una solicitud clusters.create, de la siguiente manera:
Notas:
Establece SoftwareConfig.Component en
FLINK
.De manera opcional, puedes configurar
SoftwareConfig.imageVersion
para especificar la versión de la imagen que se usará en el clúster. La versión de la imagen del clúster determina la versión del componente de Flink instalado en el clúster.La versión de la imagen debe ser 1.5 o superior para activar el componente de Flink en el clúster (consulta las versiones compatibles de Dataproc para ver las listas de las versiones de los componentes incluidas en cada versión de la imagen de Dataproc).
La versión de la imagen debe ser [TBD] o superior para ejecutar trabajos de Flink a través de la API de trabajos de Dataproc (consulta Ejecuta trabajos de Flink de Dataproc).
Establece EndpointConfig.enableHttpPortAccess en
true
para habilitar el vínculo de la puerta de enlace de componentes a la IU del servidor de historial de Flink. Habilitar la puerta de enlace de componentes también habilita el acceso a la interfaz web del administrador de trabajos de Fusion que se ejecuta en el clúster de Fusion.De manera opcional, puedes configurar
SoftwareConfig.properties
para especificar una o más propiedades del clúster.- Puedes especificar las propiedades de Flink que funcionarán como predeterminadas para las aplicaciones de Flink que ejecutes en el clúster. Por ejemplo, puedes configurar el
flink:historyserver.archive.fs.dir
para especificar la ubicación de Cloud Storage para escribir archivos de historial de trabajos de Flink (el servidor de historial de Flink que se ejecuta en el clúster de Flink usará esta ubicación).
- Puedes especificar las propiedades de Flink que funcionarán como predeterminadas para las aplicaciones de Flink que ejecutes en el clúster. Por ejemplo, puedes configurar el
De manera opcional, puedes configurar lo siguiente:
GceClusterConfig.metadata
. Por ejemplo, para especificarflink-start-yarn-session
true
con el objetivo de ejecutar el daemon de YARN de Flink (/usr/bin/flink-yarn-daemon
) en segundo plano, en el nodo instancia principal del clúster para iniciar una sesión de YARN de Flink (consulta Modo de sesión de Flink).- GceClusterConfig.serviceAccountScopes a
https://www.googleapis.com/auth/cloud-platform
(permisocloud-platform
) cuando se usan versiones de imágenes 2.0 o anteriores para habilitar el acceso de tu clúster a las APIs de Google Cloud (consulta las Prácticas recomendadas de permisos). El permisocloud-platform
se habilita de forma predeterminada cuando creas un clúster que usa la versión 2.1 o posterior con la imagen de Dataproc.
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 de Flink Job Manager que se ejecuta en el clúster de Flink . - Crear 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 mediante el recurso Jobs
de Dataproc desde la consola de Google Cloud, Google Cloud CLI o la API de Dataproc.
Console
Para enviar un trabajo de muestra de conteo de palabras de Flink desde la consola, sigue estos pasos:
Abre la página Submit a job de Dataproc en la consola de Google Cloud en tu navegador.
Completa los campos de la página Enviar un trabajo:
- En Clúster (Cluster), selecciona el nombre del clúster que quieres elegir de la lista.
- Configura Tipo de trabajo como
Flink
. - Establece Clase principal o jar (Main class or jar) en
org.apache.flink.examples.java.wordcount.WordCount
. - Configura Archivos JAR como
file:///usr/lib/flink/examples/batch/WordCount.jar
.file:///
denota un archivo ubicado en el clúster. Dataproc instaló elWordCount.jar
cuando creó el clúster de Flink.- Este campo también acepta una ruta de acceso de Cloud Storage (
gs://BUCKET/JARFILE
) o una ruta de acceso del sistema de archivos distribuido de Hadoop (HDFS) (hdfs://PATH_TO_JAR
).
Haz clic en Enviar.
- El resultado del controlador del trabajo se muestra en la página Detalles del trabajo.
- Los trabajos de Flink se enumeran en la página Trabajos de Dataproc en la consola de Google Cloud.
- Haz clic en Detener o Borrar en la página Trabajos o Detalles del trabajo para detener o borrar un trabajo.
gcloud
Para enviar un trabajo de Flink a un clúster de Dataproc Flink, ejecuta el comando gcloud dataproc jobs submit de gcloud CLI de forma local en una ventana de terminal o en Cloud Shell.
gcloud dataproc jobs submit flink \ --cluster=CLUSTER_NAME \ --region=REGION \ --class=MAIN_CLASS \ --jar=JAR_FILE \ -- JOB_ARGS
Notas:
- CLUSTER_NAME: especifica el nombre del clúster de Dataproc Flink al que se enviará el trabajo.
- REGION: Especifica una región de Compute Engine en la que se encuentre el clúster.
- MAIN_CLASS: Especifica la clase
main
de tu aplicación de Flink, por ejemplo:org.apache.flink.examples.java.wordcount.WordCount
- JAR_FILE: Especifica el archivo jar de la aplicación de Flink. Puedes especificar lo siguiente:
- Un archivo jar instalado en el clúster con el prefijo
file:///`:
file:///usr/lib/flink/examples/streaming/TopSpeedWindowing.jar
file:///usr/lib/flink/examples/batch/WordCount.jar
- Un archivo jar en Cloud Storage:
gs://BUCKET/JARFILE
- Un archivo jar en HDFS:
hdfs://PATH_TO_JAR
- Un archivo jar instalado en el clúster con el prefijo
JOB_ARGS: De manera opcional, agrega argumentos de trabajo después del guion doble (
--
).Después de enviar el trabajo, el resultado del controlador del trabajo se muestra en la terminal local o de Cloud Shell.
Program execution finished Job with JobID 829d48df4ebef2817f4000dfba126e0f has finished. Job Runtime: 13610 ms ... (after,1) (and,12) (arrows,1) (ay,1) (be,4) (bourn,1) (cast,1) (coil,1) (come,1)
REST
En esta sección, se muestra cómo enviar un trabajo de Flink a un clúster de Dataproc Flink con la API jobs.submit de Dataproc.
Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:
- PROJECT_ID: ID del proyecto de Google Cloud
- REGION: región del clúster
- CLUSTER_NAME: especifica el nombre del clúster de Dataproc Flink al que se enviará el trabajo.
Método HTTP y URL:
POST https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit
Cuerpo JSON de la solicitud:
{ "job": { "placement": { "clusterName": "CLUSTER_NAME" }, "flinkJob": { "mainClass": "org.apache.flink.examples.java.wordcount.WordCount", "jarFileUris": [ "file:///usr/lib/flink/examples/batch/WordCount.jar" ] } } }
Para enviar tu solicitud, expande una de estas opciones:
Deberías recibir una respuesta JSON similar a la que se muestra a continuación:
{ "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" }
- Los trabajos de Flink se enumeran en la página Trabajos de Dataproc en la consola de Google Cloud.
- Puedes hacer clic en Detener o Borrar en la página Trabajos o Detalles del trabajo en la consola de Google Cloud para detener o borrar un trabajo.
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 mediante la CLI de flink
.
En las siguientes secciones, se describen diferentes formas en las que puedes ejecutar un trabajo flink
de la CLI en tu clúster de Dataproc Flink.
Establece una conexión SSH al nodo principal: usa la utilidad SSH para abrir una ventana de la terminal en la VM de la instancia principal del clúster.
Configura la ruta de clase: Inicializa la ruta de clase de Hadoop desde la ventana de la terminal SSH en la VM de la instancia principal del clúster de Fusion:
export HADOOP_CLASSPATH=$(hadoop classpath)
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.
Modo de aplicación: El modo de aplicación de Flink es compatible con la versión 2.0 y posteriores de las imágenes de Dataproc. Este modo ejecuta el método
main()
del trabajo en el administrador de trabajos de YARN. El clúster se cierra después de que finaliza el trabajo.Ejemplo de envío de trabajos:
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
Cancela un trabajo en ejecución:
./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
Modo por trabajo: este modo Flink ejecuta el método
main()
del trabajo en el lado del cliente.Ejemplo de envío de trabajos:
flink run \ -m yarn-cluster \ -p 4 \ -ys 2 \ -yjm 1024m \ -ytm 2048m \ /usr/lib/flink/examples/batch/WordCount.jar
Modo de sesión: inicia una sesión de Flink YARN de larga duración y, luego, envía uno o más trabajos a la sesión.
Inicia una sesión: Puedes iniciar una sesión de Flink de una de las siguientes maneras:
Crea un clúster de Flink y agrega la marca
--metadata flink-start-yarn-session=true
al comandogcloud dataproc clusters create
(consulta Crea un clúster de Flink de Dataproc). Con esta marca habilitada, después de crear el clúster, Dataproc ejecuta/usr/bin/flink-yarn-daemon
para iniciar una sesión de Flink en el clúster.El ID de aplicación de YARN de la sesión se guarda en
/tmp/.yarn-properties-${USER}
. Puedes solicitar el ID con el comandoyarn application -list
.Ejecuta la secuencia de comandos
yarn-session.sh
de Flink, que está preinstalada en la VM de la instancia principal del clúster, con una configuración personalizada: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
Ejecuta la secuencia de comandos del wrapper
/usr/bin/flink-yarn-daemon
de Flink con la configuración predeterminada:. /usr/bin/flink-yarn-daemon
Envía un trabajo a una sesión: Ejecuta el siguiente comando para enviar un trabajo de Flip 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, incluido 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 aparece en el resultado del comando cuando inicias una sesión de Flink. Puedes ejecutar el siguiente comando para mostrar esta URL en el campoTracking-URL
:
yarn application -list -appId=<yarn-app-id> | sed 's#http://##' ```
- FLINK_MASTER_URL: Es la URL, incluido el host y el puerto, de la VM principal de Flink en la que se ejecutan los trabajos.
Quita el
Enumera los trabajos en una sesión: Para enumerar los trabajos de Flink en una sesión, realiza una de las siguientes acciones:
Ejecuta
flink list
sin argumentos. El comando busca el ID de aplicación de YARN de la sesión en/tmp/.yarn-properties-${USER}
.Obtén el ID de aplicación de YARN de la sesión desde
/tmp/.yarn-properties-${USER}
o el resultado deyarn application -list
y, luego, ejecuta<code>
flink list -yid YARN_APPLICATION_ID.Ejecuta
flink list -m FLINK_MASTER_URL
.
Detén una sesión: Para detenerla, obtén el ID de la aplicación de YARN de la sesión desde
/tmp/.yarn-properties-${USER}
o el resultado deyarn application -list
. 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:
- Trabajos de Java Beam
- 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.
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 APIs de Google Cloud desde tu clúster (consulta las Prácticas recomendadas sobre permisos). El permisocloud-platform
está habilitado de forma predeterminada (no necesitas incluir esta configuración de marca) cuando creas un clúster que usa la versión 2.1 o posterior de la imagen de Dataproc.
Usa la utilidad SSH para abrir una ventana de la terminal en el nodo instancia principal del clúster de Flink.
Inicia una sesión de Flink 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
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 , en la que se indica la compatibilidad de las 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
Empaqueta el ejemplo de conteo de palabras.
mvn package -Pflink-runner
Sube el archivo uber JAR empaquetado,
word-count-beam-bundled-0.1.jar
(~135 MB) al nodo principal del clúster de Dataproc. Puedes usargsutil cp
para transferencias de archivos más rápidas a tu clúster de Dataproc desde Cloud Storage.En tu terminal local, crea un bucket de Cloud Storage y sube el archivo uber JAR.
gsutil mb BUCKET_NAME
gsutil cp target/word-count-beam-bundled-0.1.jar gs://BUCKET_NAME/
En el nodo principal de Dataproc, descarga el archivo uber JAR.
gsutil cp gs://BUCKET_NAME/word-count-beam-bundled-0.1.jar .
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
Comprueba que los resultados se hayan escrito en tu bucket de Cloud Storage.
gsutil cat gs://BUCKET_NAME/java-wordcount-out-SHARD_ID
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 Flink Runner de Beam (también consulta la Hoja de ruta del framework 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.
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 los componentes de Apache Flink que se enumeran para las últimas y anteriores cuatro versiones de actualización de imágenes 2.0.x).--region
: Es una región de Dataproc disponible.--enable-component-gateway
: Habilita el acceso a la IU de Flink Job Manager.--scopes
: Habilita el acceso a las APIs de Google Cloud desde tu clúster (consulta las Prácticas recomendadas sobre permisos). El permisocloud-platform
está habilitado de forma predeterminada (no necesitas incluir esta configuración de marca) cuando creas un clúster que usa la versión 2.1 o posterior de la imagen de Dataproc.
Usa
gsutil
de forma local o en Cloud Shell para crear un bucket de Cloud Storage. Especificarás el BUCKET_NAME cuando ejecutes un programa de recuento de palabras de muestra.gsutil mb BUCKET_NAME
En una ventana de la terminal de la VM del clúster, inicia una sesión de Flink YARN. Toma nota de 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 el FLINK_MASTER_URL cuando ejecutes un programa de recuento de palabras de muestra.
. /usr/bin/flink-yarn-daemon
Muestra y observa la versión de Flink que ejecuta el clúster de Dataproc. Especificarás el FLINK_VERSION cuando ejecutes un programa de recuento de palabras de muestra.
flink --version
Instalar las bibliotecas de Python necesarias para el trabajo en el nodo de la instancia principal del clúster
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
Ejecuta el ejemplo de recuento de palabras en el nodo de la 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 archivo uber JAR para ejecutar el trabajo de Beam.--output
: BUCKET_NAME, creado anteriormente.
Verifica que los resultados se hayan escrito en tu bucket.
gsutil cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID
Detén la sesión de Flink en YARN.
- Obtén el ID de 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 kerberizado
El componente Flink de Dataproc admite clústeres con Kerberos. Se necesita un ticket de Kerberos válido para enviar y conservar un trabajo de Flink o iniciar un clúster de Flink. De forma predeterminada, los tickets de Kerberos son válidos durante siete días.
Acceder a la IU de Flink Job Manager
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:
- Crea un clúster de Dataproc Flink.
- Después de crear el clúster, haz clic en el vínculo de ResourceManager de YARN Puerta de enlace del componente en la pestaña Interfaz web de la página Detalles del clúster en la consola de Google Cloud.
- 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á un vínculo ApplicationMaster o History.
- 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.