Componente opcional do Flink do Managed Service for Apache Spark

O componente opcional do Apache Flink permite criar um cluster do Serviço Gerenciado para Apache Spark configurado para executar cargas de trabalho de streaming distribuído e processamento em lote. A ativação desse componente prepara o ambiente de execução para executar pipelines com estado e aplicativos de dados em grande escala.

É possível usar o cluster do Flink para executar jobs do Flink usando o recurso Jobs no console do Google Cloud , na Google Cloud CLI ou na API Dataproc, ou executar cargas de trabalho com a CLI flink no nó mestre. O componente também é compatível com jobs do Apache Beam e execução em clusters com Kerberos.

É possível usar o console do Google Cloud , a Google Cloud CLI ou a API Dataproc para criar um cluster com o componente do Flink ativado.

Recomendação: use um cluster de VM padrão de 1 mestre com o componente do Flink. Os clusters do modo de alta disponibilidade do Serviço Gerenciado para Apache Spark (com três VMs mestre) não são compatíveis com o modo de alta disponibilidade do Flink.

É possível executar jobs do Flink usando o recurso Jobs do serviço gerenciado para Apache Spark no console doGoogle Cloud , na Google Cloud CLI ou na API Dataproc.

Console

Para enviar um job de amostra de contagem de palavras do Flink pelo console:

  1. Abra a página Enviar um job do Serviço Gerenciado para Apache Spark no console doGoogle Cloud no navegador.

  2. Preencha os campos na página Enviar um job:

    1. Selecione o nome do Cluster na lista de clusters.
    2. Defina Tipo de serviço como Flink.
    3. Defina Classe principal ou jar como org.apache.flink.examples.java.wordcount.WordCount.
    4. Defina Arquivos JAR como file:///usr/lib/flink/examples/batch/WordCount.jar.
      • file:/// indica um arquivo localizado no cluster. O Serviço Gerenciado para Apache Spark instalou o WordCount.jar quando criou o cluster do Flink.
      • Esse campo também aceita um caminho do Cloud Storage (gs://BUCKET/JARFILE) ou um caminho do Hadoop Distributed File System (HDFS) (hdfs://PATH_TO_JAR).

  3. Clique em Enviar.

    • A saída do driver do job é exibida na página Detalhes do job.
    • Os jobs do Flink estão listados na página Jobs do Serviço Gerenciado para Apache Spark no console do Google Cloud .
    • Clique em Parar ou Excluir na página Jobs ou Detalhes do job para interromper ou excluir um job.

gcloud

Para enviar um job do Flink para um cluster do Flink do serviço gerenciado para Apache Spark, execute o comando da CLI gcloud gcloud dataproc jobs submit localmente em uma janela de terminal ou no Cloud Shell. 

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

Observações:

  • CLUSTER_NAME: especifique o nome do cluster do Serviço Gerenciado para Apache Spark Flink em que o job será enviado.
  • REGION: especifique uma região do Compute Engine em que o cluster está localizado.
  • MAIN_CLASS: especifique a classe main do seu aplicativo do Flink, como:
    • org.apache.flink.examples.java.wordcount.WordCount
  • JAR_FILE: especifique o arquivo JAR do aplicativo do Flink. Você pode especificar:
    • Um arquivo jar instalado no cluster, usando o prefixo file:///`:
      • file:///usr/lib/flink/examples/streaming/TopSpeedWindowing.jar
      • file:///usr/lib/flink/examples/batch/WordCount.jar
    • Um arquivo jar no Cloud Storage: gs://BUCKET/JARFILE
    • Um arquivo jar no HDFS: hdfs://PATH_TO_JAR

  • JOB_ARGS: opcionalmente, adicione argumentos de job após o traço duplo (--).

  • Depois de enviar o job, a saída do driver do job é exibida no terminal local ou do 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

Nesta seção, mostramos como enviar um job do Flink para um cluster do Serviço Gerenciado para Apache Spark Flink usando a API jobs.submit do Serviço Gerenciado para Apache Spark.

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • PROJECT_ID: Google Cloud ID do projeto
  • REGION: região do cluster
  • CLUSTER_NAME: especifique o nome do cluster do Flink do Serviço Gerenciado para Apache Spark para enviar o job.

Método HTTP e URL: 

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

Corpo JSON da solicitação:

{
  "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 a solicitação, expanda uma destas opções:

Você receberá uma resposta JSON semelhante a esta:

{
  "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"
}
  • Os jobs do Flink estão listados na página Jobs do Serviço Gerenciado para Apache Spark no console do Google Cloud .
  • Clique em Parar ou Excluir na página Jobs ou Detalhes do job no console do Google Cloud para interromper ou excluir um job.

Em vez de executar jobs do Flink usando o recurso Jobs do Serviço Gerenciado para Apache Spark, é possível executar jobs do Flink no nó do driver do cluster do Flink usando a CLI flink.

As seções a seguir descrevem diferentes maneiras de executar um job da CLI flink no cluster do Flink do Serviço Gerenciado para Apache Spark.

  1. SSH no nó mestre: use o utilitário SSH para abrir uma janela do terminal na VM mestre do cluster.

  2. Defina o caminho de classe: inicialize o caminho de classe do Hadoop na janela do terminal SSH na VM mestre do cluster do Flink:

    export HADOOP_CLASSPATH=$(hadoop classpath)

  3. Executar jobs do Flink: é possível executar jobs do Flink em diferentes modos de implantação no YARN: aplicativo, por job e modo de sessão.

    1. Modo de aplicativo: o modo de aplicativo do Flink é compatível com o Serviço Gerenciado para Apache Spark versão de imagem 2.0 e mais recentes. Esse modo executa o método main() do job no gerenciador de jobs do YARN. O cluster é encerrado após a conclusão do job.

      Exemplo de envio de job:

      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

      Listar jobs em execução:

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

      Cancelar um job em execução:

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

    2. Modo por job: esse modo do Flink executa o método main() do job no lado do cliente.

      Exemplo de envio de job:

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

    3. Modo de sessão: inicie uma sessão de longa duração do Flink YARN e envie um ou mais jobs para ela.

      1. Inicie uma sessão: é possível iniciar uma sessão do Flink de uma das seguintes maneiras:

        1. Crie um cluster do Flink adicionando a flag --metadata flink-start-yarn-session=true ao comando gcloud dataproc clusters create. Consulte Criar um cluster do Dataproc Flink. Com essa flag ativada, depois que o cluster é criado, o Serviço Gerenciado para Apache Spark executa /usr/bin/flink-yarn-daemon para iniciar uma sessão do Flink no cluster.

          O ID do aplicativo YARN da sessão é salvo em /tmp/.yarn-properties-${USER}. É possível listar o ID com o comando yarn application -list.

        2. Execute o script do Flink yarn-session.sh, que é pré-instalado na VM mestre do cluster, com configurações personalizadas:

          Exemplo com configurações personalizadas:

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

        3. Execute o script wrapper /usr/bin/flink-yarn-daemon do Flink com as configurações padrão:

          . /usr/bin/flink-yarn-daemon

      2. Envie um job para uma sessão: execute o comando a seguir para enviar um job do Flink para a sessão.

        flink run -m <var>FLINK_MASTER_URL</var>/usr/lib/flink/examples/batch/WordCount.jar
        • FLINK_MASTER_URL: o URL, incluindo host e porta, da VM mestre do Flink em que os jobs são executados. Remova o http:// prefix do URL. Esse URL é listado na resposta ao comando quando você inicia uma sessão do Flink. Execute o comando a seguir para listar esse URL no campo Tracking-URL:
        yarn application -list -appId=<yarn-app-id> | sed 's#http://##'
   ```

      3. Listar jobs em uma sessão: para listar jobs do Flink em uma sessão, siga um destes procedimentos:

        • Execute flink list sem argumentos. O comando procura o ID do aplicativo YARN da sessão em /tmp/.yarn-properties-${USER}.

        • Consiga o ID do aplicativo YARN da sessão em /tmp/.yarn-properties-${USER} ou na saída de yarn application -list e execute <code>flink list -yid YARN_APPLICATION_ID.

        • Execute flink list -m FLINK_MASTER_URL.

      4. Interromper uma sessão: para interromper a sessão, consiga o ID do aplicativo YARN da sessão em /tmp/.yarn-properties-${USER} ou na saída de yarn application -list e execute um dos seguintes comandos:

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

        yarn application -kill YARN_APPLICATION_ID

É possível executar jobs do Apache Beam no Serviço Gerenciado para Apache Spark usando o FlinkRunner.

É possível executar jobs do Beam no Flink das seguintes maneiras:

  1. Jobs do Beam em Java
  2. Jobs do Beam portáteis

Jobs do Beam em Java

Empacotar os jobs do Beam em um arquivo JAR. Forneça o arquivo JAR empacotado com as dependências necessárias para executar o job.

O exemplo a seguir executa um job do Java Beam no nó mestre do cluster do Serviço Gerenciado para Apache Spark.

  1. Crie um cluster do Serviço Gerenciado para Apache Spark com o componente do Flink ativado.

    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: a versão de imagem do cluster, que determina a versão do Flink instalada no cluster. Por exemplo, consulte as versões do componente Flink do Apache listadas para a versão mais recente e quatro anteriores versões de lançamento de imagem 2.0.x.
    • --region: uma região compatível com o Serviço Gerenciado para Apache Spark.
    • --enable-component-gateway: ativa o acesso à IU do Flink Job Manager.
    • --scopes: ative o acesso às APIs do Google Cloud pelo cluster (consulte Práticas recomendadas de escopos). O escopo cloud-platform é ativado por padrão (não é necessário incluir essa configuração de flag) quando você cria um cluster que usa a versão 2.1 ou posterior da imagem do Managed Service para Apache Spark.

  2. Use o utilitário SSH para abrir uma janela do terminal no nó mestre do cluster do Flink.

  3. Inicie uma sessão do Flink YARN no nó mestre do cluster do Serviço Gerenciado para Apache Spark.

    . /usr/bin/flink-yarn-daemon

    Anote a versão do Flink no cluster do Serviço Gerenciado para Apache Spark.

    flink --version

  4. Na máquina local, gere o exemplo de contagem de palavras canônica do Beam em Java.

    Escolha uma versão do Beam compatível com a versão do Flink no cluster do Serviço Gerenciado para Apache Spark. Consulte a tabela de compatibilidade de versões do Flink, que lista a compatibilidade de versões do Beam-Flink.

    Abra o arquivo POM gerado. Verifique a versão do executor do Flink do Beam especificada pela tag <flink.artifact.name>. Se o executor do Flink do Beam no nome do artefato do Flink não corresponder à versão do Flink no cluster, atualize o número da versão para corresponder.

    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. Crie um exemplo de contagem de palavras.

    mvn package -Pflink-runner

  6. Faça upload do arquivo JAR uber empacotado, word-count-beam-bundled-0.1.jar (~135 MB), para o nó mestre do cluster do Serviço Gerenciado para Apache Spark. Use o gcloud storage cp para transferências de arquivos mais rápidas do Cloud Storage para o cluster do Serviço Gerenciado para Apache Spark.

    1. No terminal local, crie um bucket do Cloud Storage e faça o upload do JAR uber.

      gcloud storage buckets create BUCKET_NAME

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

    2. No nó mestre do Serviço Gerenciado para Apache Spark, faça o download do uber JAR.

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

  7. Execute o job do Java Beam no nó do driver do cluster do Serviço Gerenciado para Apache Spark.

    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. Verifique se os resultados foram gravados no seu bucket do Cloud Storage.

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

  9. Interrompa a sessão YARN do Flink.

    yarn application -list

yarn application -kill YARN_APPLICATION_ID

Jobs do Beam portáteis

Para executar jobs do Beam escritos em Python, Go e outras linguagens compatíveis, use o FlinkRunner e o PortableRunner, conforme descrito na página Flink Runner do Beam (consulte também o roteiro do framework de portabilidade).

O exemplo a seguir executa um job portátil do Beam em Python no nó mestre do cluster do Serviço Gerenciado para Apache Spark.

  1. Crie um cluster do Serviço Gerenciado para Apache Spark com os componentes Flink e Docker ativados.

    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

    Observações:

    • --optional-components: Flink e Docker.
    • --image-version: a versão de imagem do cluster, que determina a versão do Flink instalada no cluster. Por exemplo, consulte as versões do componente do Apache Flink listadas para as quatro versões de lançamento de imagem 2.0.x mais recentes e anteriores.
    • --region: uma região disponível do Serviço Gerenciado para Apache Spark.
    • --enable-component-gateway: ativa o acesso à interface do gerenciador de jobs do Flink.
    • --scopes: ative o acesso às APIs do Google Cloud pelo cluster (consulte Práticas recomendadas de escopos). O escopo cloud-platform é ativado por padrão (não é necessário incluir essa configuração de flag) quando você cria um cluster que usa a versão 2.1 ou posterior da versão de imagem do Serviço Gerenciado para Apache Spark.

  2. Use a CLI gcloud localmente ou no Cloud Shell para criar um bucket do Cloud Storage. Você vai especificar o BUCKET_NAME ao executar um programa de exemplo de contagem de palavras.

    gcloud storage buckets create BUCKET_NAME

  3. Em uma janela de terminal na VM do cluster, inicie uma sessão do Flink YARN. Anote o URL mestre do Flink, o endereço do mestre do Flink em que os jobs são executados. Você vai especificar o FLINK_MASTER_URL ao executar um programa de contagem de palavras de amostra.

    . /usr/bin/flink-yarn-daemon

    Exiba e anote a versão do Flink que executa o cluster do Serviço Gerenciado para Apache Spark. Você vai especificar o FLINK_VERSION ao executar um programa de contagem de palavras de amostra.

    flink --version

  4. Instale as bibliotecas Python necessárias para o job no nó mestre do cluster.

  5. Instale uma versão do Beam compatível com a versão do Flink no cluster.

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

  6. Execute o exemplo de contagem de palavras no nó mestre do cluster.

    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

    Observações:

    • --runner: FlinkRunner.
    • --flink_version: FLINK_VERSION, observado anteriormente.
    • --flink_master: FLINK_MASTER_URL, observado anteriormente.
    • --flink_submit_uber_jar: use o uber JAR para executar o job do Beam.
    • --output: BUCKET_NAME, criado anteriormente.

  7. Verifique se os resultados foram gravados no bucket.

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

  8. Interrompa a sessão YARN do Flink.

    1. Encontre o ID do aplicativo.
    yarn application -list

1. Insert the <var>YARN_APPLICATION_ID</var>, then stop the session.

yarn application -kill

O componente do Serviço Gerenciado para Apache Spark Flink é compatível com clusters Kerberizados. É necessário um tíquete Kerberos válido para enviar e manter um job do Flink ou iniciar um cluster do Flink. Por padrão, um tíquete do Kerberos permanece válido por sete dias.

A interface da Web do Flink Job Manager está disponível durante a execução de um job de Flink ou um cluster de sessão do Flink. Para usar a interface da Web:

  1. Criar um cluster do Flink do Serviço Gerenciado para Apache Spark.
  2. Após a criação do cluster, clique no link do ResourceManager do YARN do Gateway de componentes na guia "Interface da Web" da página Detalhes do cluster no console do Google Cloud .
  3. Na IU do YARN Resource Manager, identifique a entrada do aplicativo de cluster Flink. Dependendo do status de conclusão de um job, um link ApplicationMaster ou Histórico será listado.
  4. Para um job de streaming de longa duração, clique no link ApplicationManager para abrir o painel do Flink. Para um job concluído, clique no link Histórico para visualizar os detalhes dele.