Componente Flink facoltativo di Dataproc

Puoi attivare componenti aggiuntivi come Flink quando crei un cluster Dataproc utilizzando la funzionalità Componenti facoltativi. Questa pagina mostra come creare un cluster Dataproc con il componente facoltativo Apache Flink attivato (un cluster Flink), quindi eseguire i job Flink sul cluster.

Puoi utilizzare il tuo cluster Flink per:

1. Esegui job Flink utilizzando la Jobsrisorsa Dataproc dalla Google Cloud console, da Google Cloud CLI o dall'API Dataproc.

2. Esegui job Flink utilizzando l'interfaccia a riga di comando flink in esecuzione sul nodo master del cluster Flink.

3. Esegui job Apache Beam su Flink

4. Esegui Flink su un cluster Kerberizzato

Crea un cluster Dataproc Flink

Puoi utilizzare la console Google Cloud , Google Cloud CLI o l'API Dataproc per creare un cluster Dataproc con il componente Flink attivato nel cluster.

Suggerimento:utilizza un cluster di VM standard con un master con il componente Flink. I cluster in modalità ad alta disponibilità di Dataproc (con 3 VM master) non supportano la modalità ad alta disponibilità di Flink.

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

Dopo aver creato un cluster Flink

• Utilizza il link Flink History Server in Component Gateway per visualizzare Flink History Server in esecuzione sul cluster Flink.
• Utilizza YARN ResourceManager link in Component Gateway per visualizzare l'interfaccia web di Flink Job Manager in esecuzione sul cluster Flink .
• Crea un server di cronologia permanente Dataproc per visualizzare i file della cronologia dei job Flink scritti dai cluster Flink esistenti ed eliminati.

Esegui job Flink utilizzando la risorsa Dataproc Jobs

Puoi eseguire job Flink utilizzando la risorsa Dataproc Jobs dalla consoleGoogle Cloud , da Google Cloud CLI o dall'API 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"
}

Esegui job Flink utilizzando l'interfaccia a riga di comando flink

Anziché eseguire job Flink utilizzando la risorsa Dataproc Jobs, puoi eseguire job Flink sul nodo master del cluster Flink utilizzando la CLI flink.

Le seguenti sezioni descrivono diversi modi per eseguire un job CLI flink sul cluster Dataproc Flink.

1. Accedi al nodo master tramite SSH:utilizza l'utilità SSH per aprire una finestra del terminale sulla VM master del cluster.

2. Imposta il classpath:inizializza il classpath di Hadoop dalla finestra del terminale SSH nella VM master del cluster Flink:

   export HADOOP_CLASSPATH=$(hadoop classpath)
   

3. Esegui job Flink:puoi eseguire job Flink in diverse modalità di deployment su YARN: modalità applicazione, per job e sessione.

   1. Modalità applicazione:la modalità applicazione di Flink è supportata da Dataproc versione 2.0 e successive. Questa modalità esegue il metodo main() del job su YARN Job Manager. Il cluster si arresta al termine del job.

      Esempio di invio del 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
      

      Elenca i job in esecuzione:

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

      Annulla un job in esecuzione:

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

   2. Modalità per job:questa modalità Flink esegue il metodo main() del job sul lato client.

      Esempio di invio del job:

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

   3. Modalità sessione: avvia una sessione Flink YARN a lunga esecuzione, quindi invia uno o più job alla sessione.

      1. Avvia una sessione:puoi avviare una sessione Flink in uno dei seguenti modi:

         1. Crea un cluster Flink aggiungendo il flag --metadata flink-start-yarn-session=true al comando gcloud dataproc clusters create (vedi Creare un cluster Dataproc Flink). Se questo flag è abilitato, dopo la creazione del cluster, Dataproc esegue /usr/bin/flink-yarn-daemon per avviare una sessione Flink sul cluster.

            L'ID applicazione YARN della sessione viene salvato in /tmp/.yarn-properties-${USER}. Puoi elencare l'ID con il comando yarn application -list.

         2. Esegui lo script Flink yarn-session.sh, preinstallato sulla VM master del cluster, con impostazioni personalizzate:

            Esempio con impostazioni personalizzate:

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

         3. Esegui lo script wrapper /usr/bin/flink-yarn-daemon di Flink con le impostazioni predefinite:

            . /usr/bin/flink-yarn-daemon
            

      2. Invia un job a una sessione: esegui questo comando per inviare un job Flink alla sessione.

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

         • FLINK_MASTER_URL: l'URL, inclusi host e porta, della VM master Flink in cui vengono eseguiti i job. Rimuovi http:// prefix dall'URL. Questo URL è elencato nell'output del comando quando avvii una sessione di Flink. Puoi eseguire questo comando per elencare questo URL nel campo Tracking-URL:

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

      3. Elencare i job in una sessione:per elencare i job Flink in una sessione, esegui una delle seguenti operazioni:

         • Esegui flink list senza argomenti. Il comando cerca l'ID applicazione YARN della sessione in /tmp/.yarn-properties-${USER}.

         • Ottieni l'ID applicazione YARN della sessione da /tmp/.yarn-properties-${USER} o dall'output di yarn application -list, quindi esegui <code>flink list -yid YARN_APPLICATION_ID.

         • Esegui flink list -m FLINK_MASTER_URL.

      4. Interrompi una sessione: per interrompere la sessione, ottieni l'ID applicazione YARN della sessione da /tmp/.yarn-properties-${USER} o dall'output di yarn application -list, quindi esegui uno dei seguenti comandi:

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

         yarn application -kill YARN_APPLICATION_ID
         

Eseguire job Apache Beam su Flink

Puoi eseguire job Apache Beam su Dataproc utilizzando FlinkRunner.

Puoi eseguire i job Beam su Flink nei seguenti modi:

1. Job Java Beam
2. Job Portable Beam

Job Java Beam

Pacchettizza i job Beam in un file JAR. Fornisci il file JAR in bundle con le dipendenze necessarie per eseguire il job.

L'esempio seguente esegue un job Java Beam dal nodo master del cluster Dataproc.

1. Crea un cluster Dataproc con il componente Flink abilitato.

   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: la versione dell'immagine del cluster, che determina la versione di Flink installata sul cluster (ad esempio, vedi le versioni dei componenti Apache Flink elencate per le quattro versioni di rilascio dell'immagine 2.0.x più recenti e precedenti).
   • --region: una regione Dataproc supportata.
   • --enable-component-gateway: abilita l'accesso all'interfaccia utente di Flink Job Manager.
   • --scopes: abilita l'accesso alle API da parte del cluster (vedi Best practice per gli ambiti). Google Cloud L'ambito cloud-platform è abilitato per impostazione predefinita (non è necessario includere questa impostazione del flag) quando crei un cluster che utilizza la versione immagine Dataproc 2.1 o successive.

2. Utilizza l'utilità SSH per aprire una finestra del terminale sul nodo master del cluster Flink.

3. Avvia una sessione Flink YARN sul nodo master del cluster Dataproc.

   . /usr/bin/flink-yarn-daemon
   

   Prendi nota della versione di Flink sul cluster Dataproc.

   flink --version
   

4. Sulla tua macchina locale, genera l'esempio canonico di conteggio delle parole di Beam in Java.

   Scegli una versione di Beam compatibile con la versione di Flink sul tuo cluster Dataproc. Consulta la tabella Compatibilità delle versioni di Flink che elenca la compatibilità delle versioni di Beam-Flink.

   Apri il file POM generato. Controlla la versione del runner Beam Flink specificata dal tag <flink.artifact.name>. Se la versione del runner Beam Flink nel nome dell'artefatto Flink non corrisponde alla versione di Flink sul cluster, aggiorna il numero di versione in modo che corrisponda.

   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. Pacchettizza l'esempio di conteggio parole.

   mvn package -Pflink-runner
   

6. Carica il file JAR uber compresso, word-count-beam-bundled-0.1.jar (~135 MB) nel nodo master del cluster Dataproc. Puoi utilizzare gcloud storage cp per trasferire più rapidamente i file al cluster Dataproc da Cloud Storage.

   1. Nel terminale locale, crea un bucket Cloud Storage e carica l'uber JAR.

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

   2. Sul nodo master di Dataproc, scarica l'uber JAR.

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

7. Esegui il job Java Beam sul nodo master del cluster 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. Verifica che i risultati siano stati scritti nel bucket Cloud Storage.

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

9. Interrompi la sessione Flink YARN.

   yarn application -list
   
   yarn application -kill YARN_APPLICATION_ID
   

Portable Beam Jobs

Per eseguire job Beam scritti in Python, Go e altri linguaggi supportati, puoi utilizzare FlinkRunner e PortableRunner come descritto nella pagina Flink Runner di Beam (vedi anche Roadmap del framework di portabilità).

L'esempio seguente esegue un job Beam portatile in Python dal nodo master del cluster Dataproc.

1. Crea un cluster Dataproc con i componenti Flink e Docker abilitati.

   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
   

   Note:

   • --optional-components: Flink e Docker.
   • --image-version: la versione dell'immagine del cluster, che determina la versione di Flink installata sul cluster (ad esempio, vedi le versioni dei componenti Apache Flink elencate per le quattro versioni di rilascio dell'immagine 2.0.x più recenti e precedenti).
   • --region: una regione Dataproc disponibile.
   • --enable-component-gateway: attiva l'accesso all'interfaccia utente di Flink Job Manager.
   • --scopes: attiva l'accesso alle API da parte del cluster (vedi Best practice per gli ambiti). Google Cloud L'ambito cloud-platform è abilitato per impostazione predefinita (non è necessario includere questa impostazione del flag) quando crei un cluster che utilizza la versione immagine Dataproc 2.1 o successive.

2. Utilizza gcloud CLI localmente o in Cloud Shell per creare un bucket Cloud Storage. Specificherai BUCKET_NAME quando esegui un programma di conteggio delle parole di esempio.

   gcloud storage buckets create BUCKET_NAME
   

3. In una finestra del terminale sulla VM del cluster, avvia una sessione Flink YARN. Prendi nota dell'URL master di Flink, l'indirizzo del master di Flink in cui vengono eseguiti i job. Specificherai il FLINK_MASTER_URL quando esegui un programma di conteggio delle parole di esempio.

   . /usr/bin/flink-yarn-daemon
   

   Visualizza e prendi nota della versione di Flink in esecuzione nel cluster Dataproc. Specificherai il FLINK_VERSION quando esegui un programma di conteggio delle parole di esempio.

   flink --version
   

4. Installa le librerie Python necessarie per il job sul nodo master del cluster.

5. Installa una versione di Beam compatibile con la versione di Flink sul cluster.

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

6. Esegui l'esempio di conteggio delle parole sul nodo master del 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
   

   Note:

   • --runner: FlinkRunner.
   • --flink_version: FLINK_VERSION, come indicato in precedenza.
   • --flink_master: FLINK_MASTER_URL, come indicato in precedenza.
   • --flink_submit_uber_jar: utilizza il file JAR uber per eseguire il job Beam.
   • --output: BUCKET_NAME, creato in precedenza.

7. Verifica che i risultati siano stati scritti nel bucket.

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

8. Interrompi la sessione Flink YARN.

   1. Recupera l'ID applicazione.

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

Esegui Flink su un cluster Kerberizzato

Il componente Dataproc Flink supporta i cluster Kerberizzati. Per inviare e rendere persistente un job Flink o per avviare un cluster Flink è necessario un ticket Kerberos valido. Per impostazione predefinita, un ticket Kerberos rimane valido per sette giorni.

Accedere all'interfaccia utente di Flink Job Manager

L'interfaccia web di Flink Job Manager è disponibile durante l'esecuzione di un job Flink o di un cluster di sessione Flink. Per utilizzare l'interfaccia web:

1. Crea un cluster Dataproc Flink.
2. Dopo la creazione del cluster, fai clic sul link YARN ResourceManager di Component Gateway nella scheda Interfaccia web della pagina Dettagli cluster nella console Google Cloud .
3. Nell'interfaccia utente di YARN Resource Manager, identifica la voce dell'applicazione del cluster Flink. A seconda dello stato di completamento di un job, verrà visualizzato un link ApplicationMaster o History.

   

4. Per un job di streaming a esecuzione prolungata, fai clic sul link ApplicationManager per aprire la dashboard di Flink; per un job completato, fai clic sul link Cronologia per visualizzare i dettagli del job.