Insérer un sujet Kafka dans Hive

Apache Kafka est une plate-forme de streaming distribuée Open Source pour les pipelines de données et l'intégration de données en temps réel. Il fournit un système de streaming efficace et évolutif pour une utilisation dans diverses applications, y compris :

  • Analyse en temps réel
  • Traitement par flux
  • Agrégation de journaux
  • Messagerie distribuée
  • Streaming d'événements

Objectifs

  1. Installez Kafka sur un cluster Dataproc à haute disponibilité avec ZooKeeper (appelé "cluster Dataproc Kafka" dans ce tutoriel).

  2. Créez des données client fictives, puis publiez-les dans un sujet Kafka.

  3. Créez des tables Hive Parquet et ORC dans Cloud Storage pour recevoir les données de sujet Kafka en flux continu.

  4. Envoyez une tâche PySpark pour vous abonner au sujet Kafka et le diffuser dans Cloud Storage au format Parquet et ORC.

  5. Exécutez une requête sur les données de la table Hive diffusées pour compter les messages Kafka diffusés.

Coûts

Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :

Pour obtenir une estimation des coûts en fonction de votre utilisation prévue, utilisez le simulateur de coût.

Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai gratuit.

Une fois que vous avez terminé les tâches décrites dans ce document, vous pouvez éviter de continuer à payer des frais en supprimant les ressources que vous avez créées. Pour en savoir plus, consultez la section Effectuer un nettoyage.

Avant de commencer

Si vous ne l'avez pas déjà fait, créez un projet Google Cloud .

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Enable the APIs

    8.
  8. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  9. Click Create.
  10. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
    1. In the Get started section, do the following:
      • Enter a globally unique name that meets the bucket naming requirements.
      • To add a bucket label, expand the Labels section (), click Add label, and specify a key and a value for your label.
    2. In the Choose where to store your data section, do the following:
      1. Select a Location type.
      2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
      3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

        Set up cross-bucket replication

        1. In the Bucket menu, select a bucket.

        2. In the Replication settings section, click Configure to configure settings for the replication job.

          The Configure cross-bucket replication pane appears.

          • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click Add a prefix.
          • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
          • Click Done.
    3. In the Choose how to store your data section, do the following:
      1. Select a default storage class for the bucket or Autoclass for automatic storage class management of your bucket's data.
      2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
    4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
    5. In the Choose how to protect object data section, do the following:
      • Select any of the options under Data protection that you want to set for your bucket.
        • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
        • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
        • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
          • To enable Object Retention Lock, click the Enable object retention checkbox.
          • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
      • To choose how your object data will be encrypted, expand the Data encryption section (), and select a Data encryption method.
  11. Click Create.

    12. Étapes du tutoriel

    Suivez les étapes ci-dessous pour créer un cluster Kafka Dataproc afin de lire un sujet Kafka dans Cloud Storage au format Parquet ou ORC.

    Copier le script d'installation de Kafka dans Cloud Storage

    Le script d'kafka.sh action d'initialisation installe Kafka sur un cluster Dataproc.

    1. Parcourez le code.

      #!/bin/bash
#    Copyright 2015 Google, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#
# This script installs Apache Kafka (http://kafka.apache.org) on a Google Cloud
# Dataproc cluster.

set -euxo pipefail

readonly ZOOKEEPER_HOME=/usr/lib/zookeeper
readonly KAFKA_HOME=/usr/lib/kafka
readonly KAFKA_PROP_FILE='/etc/kafka/conf/server.properties'
readonly ROLE="$(/usr/share/google/get_metadata_value attributes/dataproc-role)"
readonly RUN_ON_MASTER="$(/usr/share/google/get_metadata_value attributes/run-on-master || echo false)"
readonly KAFKA_ENABLE_JMX="$(/usr/share/google/get_metadata_value attributes/kafka-enable-jmx || echo false)"
readonly KAFKA_JMX_PORT="$(/usr/share/google/get_metadata_value attributes/kafka-jmx-port || echo 9999)"
readonly INSTALL_KAFKA_PYTHON="$(/usr/share/google/get_metadata_value attributes/install-kafka-python || echo false)"

# The first ZooKeeper server address, e.g., "cluster1-m-0:2181".
ZOOKEEPER_ADDRESS=''
# Integer broker ID of this node, e.g., 0
BROKER_ID=''

function retry_apt_command() {
  cmd="$1"
  for ((i = 0; i < 10; i++)); do
    if eval "$cmd"; then
      return 0
    fi
    sleep 5
  done
  return 1
}

function recv_keys() {
  retry_apt_command "apt-get install -y gnupg2 &&\
                     apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
}

function update_apt_get() {
  retry_apt_command "apt-get update"
}

function install_apt_get() {
  pkgs="$@"
  retry_apt_command "apt-get install -y $pkgs"
}

function err() {
  echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $@" >&2
  return 1
}

# Returns the list of broker IDs registered in ZooKeeper, e.g., " 0, 2, 1,".
function get_broker_list() {
  ${KAFKA_HOME}/bin/zookeeper-shell.sh "${ZOOKEEPER_ADDRESS}" \
    <<<"ls /brokers/ids" |
    grep '\[.*\]' |
    sed 's/\[/ /' |
    sed 's/\]/,/'
}

# Waits for zookeeper to be up or time out.
function wait_for_zookeeper() {
  for i in {1..20}; do
    if "${ZOOKEEPER_HOME}/bin/zkCli.sh" -server "${ZOOKEEPER_ADDRESS}" ls /; then
      return 0
    else
      echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}, retry ${i}..."
      sleep 5
    fi
  done
  echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}" >&2
  exit 1
}

# Wait until the current broker is registered or time out.
function wait_for_kafka() {
  for i in {1..20}; do
    local broker_list=$(get_broker_list || true)
    if [[ "${broker_list}" == *" ${BROKER_ID},"* ]]; then
      return 0
    else
      echo "Kafka broker ${BROKER_ID} is not registered yet, retry ${i}..."
      sleep 5
    fi
  done
  echo "Failed to start Kafka broker ${BROKER_ID}." >&2
  exit 1
}

function install_and_configure_kafka_server() {
  # Find zookeeper list first, before attempting any installation.
  local zookeeper_client_port
  zookeeper_client_port=$(grep 'clientPort' /etc/zookeeper/conf/zoo.cfg |
    tail -n 1 |
    cut -d '=' -f 2)

  local zookeeper_list
  zookeeper_list=$(grep '^server\.' /etc/zookeeper/conf/zoo.cfg |
    cut -d '=' -f 2 |
    cut -d ':' -f 1 |
    sort |
    uniq |
    sed "s/$/:${zookeeper_client_port}/" |
    xargs echo |
    sed "s/ /,/g")

  if [[ -z "${zookeeper_list}" ]]; then
    # Didn't find zookeeper quorum in zoo.cfg, but possibly workers just didn't
    # bother to populate it. Check if YARN HA is configured.
    zookeeper_list=$(bdconfig get_property_value --configuration_file \
      /etc/hadoop/conf/yarn-site.xml \
      --name yarn.resourcemanager.zk-address 2>/dev/null)
  fi

  # If all attempts failed, error out.
  if [[ -z "${zookeeper_list}" ]]; then
    err 'Failed to find configured Zookeeper list; try "--num-masters=3" for HA'
  fi

  ZOOKEEPER_ADDRESS="${zookeeper_list%%,*}"

  # Install Kafka from Dataproc distro.
  install_apt_get kafka-server || dpkg -l kafka-server ||
    err 'Unable to install and find kafka-server.'

  mkdir -p /var/lib/kafka-logs
  chown kafka:kafka -R /var/lib/kafka-logs

  if [[ "${ROLE}" == "Master" ]]; then
    # For master nodes, broker ID starts from 10,000.
    if [[ "$(hostname)" == *-m ]]; then
      # non-HA
      BROKER_ID=10000
    else
      # HA
      BROKER_ID=$((10000 + $(hostname | sed 's/.*-m-\([0-9]*\)$/\1/g')))
    fi
  else
    # For worker nodes, broker ID is a random number generated less than 10000.
    # 10000 is choosen since the max broker ID allowed being set is 10000.
    BROKER_ID=$((RANDOM % 10000))
  fi
  sed -i 's|log.dirs=/tmp/kafka-logs|log.dirs=/var/lib/kafka-logs|' \
    "${KAFKA_PROP_FILE}"
  sed -i 's|^\(zookeeper\.connect=\).*|\1'${zookeeper_list}'|' \
    "${KAFKA_PROP_FILE}"
  sed -i 's,^\(broker\.id=\).*,\1'${BROKER_ID}',' \
    "${KAFKA_PROP_FILE}"
  echo -e '\nreserved.broker.max.id=100000' >>"${KAFKA_PROP_FILE}"
  echo -e '\ndelete.topic.enable=true' >>"${KAFKA_PROP_FILE}"

  if [[ "${KAFKA_ENABLE_JMX}" == "true" ]]; then
    sed -i '/kafka-run-class.sh/i export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Djava.net.preferIPv4Stack=true"' /usr/lib/kafka/bin/kafka-server-start.sh
    sed -i "/kafka-run-class.sh/i export JMX_PORT=${KAFKA_JMX_PORT}" /usr/lib/kafka/bin/kafka-server-start.sh
  fi

  wait_for_zookeeper

  # Start Kafka.
  service kafka-server restart

  wait_for_kafka
}

function install_kafka_python_package() {
  KAFKA_PYTHON_PACKAGE="kafka-python==2.0.2"
  if [[ "${INSTALL_KAFKA_PYTHON}" != "true" ]]; then
    return
  fi

  if [[ "$(echo "${DATAPROC_IMAGE_VERSION} > 2.0" | bc)" -eq 1 ]]; then
    /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}"; }
  else
    OS=$(. /etc/os-release && echo "${ID}")
    if [[ "${OS}" == "rocky" ]]; then
      yum install -y python2-pip
    else
      apt-get install -y python-pip
    fi
    pip2 install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; pip2 install "${KAFKA_PYTHON_PACKAGE}"; } || { sleep 10; pip install "${KAFKA_PYTHON_PACKAGE}"; }
  fi
}

function remove_old_backports {
  # This script uses 'apt-get update' and is therefore potentially dependent on
  # backports repositories which have been archived.  In order to mitigate this
  # problem, we will remove any reference to backports repos older than oldstable

  # https://github.com/GoogleCloudDataproc/initialization-actions/issues/1157
  oldstable=$(curl -s https://deb.debian.org/debian/dists/oldstable/Release | awk '/^Codename/ {print $2}');
  stable=$(curl -s https://deb.debian.org/debian/dists/stable/Release | awk '/^Codename/ {print $2}');

  matched_files="$(grep -rsil '\-backports' /etc/apt/sources.list*)"
  if [[ -n "$matched_files" ]]; then
    for filename in "$matched_files"; do
      grep -e "$oldstable-backports" -e "$stable-backports" "$filename" || \
        sed -i -e 's/^.*-backports.*$//' "$filename"
    done
  fi
}

function main() {
  OS=$(. /etc/os-release && echo "${ID}")
  if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} <= 2.1" | bc -l) == 1 ]]; then
    remove_old_backports
  fi
  recv_keys || err 'Unable to receive keys.'
  update_apt_get || err 'Unable to update packages lists.'
  install_kafka_python_package

  # Only run the installation on workers; verify zookeeper on master(s).
  if [[ "${ROLE}" == 'Master' ]]; then
    service zookeeper-server status ||
      err 'Required zookeeper-server not running on master!'
    if [[ "${RUN_ON_MASTER}" == "true" ]]; then
      # Run installation on masters.
      install_and_configure_kafka_server
    else
      # On master nodes, just install kafka command-line tools and libs but not
      # kafka-server.
      install_apt_get kafka ||
        err 'Unable to install kafka libraries on master!'
    fi
  else
    # Run installation on workers.
    install_and_configure_kafka_server
  fi
}

main

    2. Copiez le script d'kafka.sh action d'initialisation dans votre bucket Cloud Storage. Ce script installe Kafka sur un cluster Dataproc.

      1. Ouvrez Cloud Shell, puis exécutez la commande suivante : 

        gcloud storage cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/

        Effectuez les remplacements suivants :

        • REGION : kafka.sh est stocké dans des buckets publics tagués par région dans Cloud Storage. Spécifiez une région Compute Engine géographiquement proche (par exemple, us-central1).
        • BUCKET_NAME : nom du bucket Cloud Storage.

    Créer un cluster Dataproc Kafka

    1. Ouvrez Cloud Shell, puis exécutez la commande gcloud dataproc clusters create suivante pour créer un cluster à haute disponibilité Dataproc qui installe les composants Kafka et ZooKeeper :

      gcloud dataproc clusters create KAFKA_CLUSTER \
    --project=PROJECT_ID \
    --region=REGION \
    --image-version=2.1-debian11 \
    --num-masters=3 \
    --enable-component-gateway \
    --initialization-actions=gs://BUCKET_NAME/scripts/kafka.sh

      Remarques :

      • KAFKA_CLUSTER : nom du cluster, qui doit être unique dans un projet. Le nom doit commencer par une lettre minuscule et peut contenir jusqu'à 51 lettres minuscules, chiffres et traits d'union. Il ne peut pas se terminer par un trait d'union. Le nom d'un cluster supprimé peut être réutilisé.
      • PROJECT_ID : projet à associer à ce cluster.
      • REGION : région Compute Engine dans laquelle le cluster sera situé, par exemple us-central1.
        • Vous pouvez ajouter l'option facultative --zone=ZONE pour spécifier une zone dans la région indiquée, par exemple us-central1-a. Si vous ne spécifiez pas de zone, la fonctionnalité de sélection de zone automatique de Dataproc en choisit une dans la région spécifiée.
      • --image-version : la version d'image Dataproc 2.1-debian11 est recommandée pour ce tutoriel. Remarque : Chaque version d'image contient un ensemble de composants préinstallés, y compris le composant Hive utilisé dans ce tutoriel (voir Versions d'image Dataproc compatibles).
      • --num-master : 3 nœuds maîtres créent un cluster à haute disponibilité. Le composant Zookeeper, requis par Kafka, est préinstallé sur un cluster à haute disponibilité.
      • --enable-component-gateway : active la passerelle des composants Dataproc.
      • BUCKET_NAME : nom de votre bucket Cloud Storage contenant le script d'initialisation /scripts/kafka.sh (voir Copier le script d'installation de Kafka dans Cloud Storage).

    Créer un sujet custdata Kafka

    Pour créer un sujet Kafka sur le cluster Kafka Dataproc :

    1. Utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur la VM maître du cluster.

    2. Créez un sujet custdata Kafka.

      /usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --create --topic custdata

      Remarques :

      • KAFKA_CLUSTER : insérez le nom de votre cluster Kafka. -w-0:9092 désigne le courtier Kafka exécuté sur le port 9092 du nœud worker-0.

      • Vous pouvez exécuter les commandes suivantes après avoir créé le sujet custdata :

        # List all topics.
/usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --list

        
# Consume then display topic data.
/usr/lib/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --topic custdata

        
# Count the number of messages in the topic.
/usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
    --broker-list KAFKA_CLUSTER-w-0:9092 \
    --topic custdata

        
# Delete topic.
/usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --delete --topic custdata

    Publier du contenu dans le sujet Kafka custdata

    Le script suivant utilise l'outil Kafka kafka-console-producer.sh pour générer des données client fictives au format CSV.

    1. Copiez le script, puis collez-le dans le terminal SSH sur le nœud maître de votre cluster Kafka. Appuyez sur <return> pour exécuter le script.

      for i in {1..10000}; do \
custname="cust name${i}"
uuid=$(dbus-uuidgen)
age=$((45 + $RANDOM % 45))
amount=$(echo "$(( $RANDOM % 99999 )).$(( $RANDOM % 99 ))")
message="${uuid}:${custname},${age},${amount}"
echo ${message}
done | /usr/lib/kafka/bin/kafka-console-producer.sh \
--broker-list KAFKA_CLUSTER-w-0:9092 \
--topic custdata \
--property "parse.key=true" \
--property "key.separator=:"

      Remarques :

      • KAFKA_CLUSTER : nom de votre cluster Kafka.

    2. Exécutez la commande Kafka suivante pour vérifier que le sujet custdata contient 10 000 messages.

      /usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list KAFKA_CLUSTER-w-0:9092 \
--topic custdata

      Remarques :

      • KAFKA_CLUSTER : nom de votre cluster Kafka.

      Résultat attendu : 

      custdata:0:10000

    Créer des tables Hive dans Cloud Storage

    Créez des tables Hive pour recevoir les données de sujet Kafka diffusées. Pour créer des tables Hive cust_parquet (Parquet) et cust_orc (ORC) dans votre bucket Cloud Storage, procédez comme suit.

    1. Insérez votre BUCKET_NAME dans le script suivant, puis copiez-le et collez-le dans le terminal SSH sur le nœud maître de votre cluster Kafka. Appuyez ensuite sur <return> pour créer un script ~/hivetables.hql (Hive Query Language).

      Vous exécuterez le script ~/hivetables.hql à l'étape suivante pour créer des tables Hive Parquet et ORC dans votre bucket Cloud Storage.

      cat > ~/hivetables.hql <<EOF
drop table if exists cust_parquet;
create external table if not exists cust_parquet
(uuid string, custname string, age string, amount string)
row format delimited fields terminated by ','
stored as parquet
location "gs://BUCKET_NAME/tables/cust_parquet";

      
drop table if exists cust_orc;
create external table if not exists cust_orc
(uuid string, custname string, age string, amount string)
row format delimited fields terminated by ','
stored as orc
location "gs://BUCKET_NAME/tables/cust_orc";
EOF

    2. Dans le terminal SSH du nœud maître de votre cluster Kafka, envoyez le job Hive ~/hivetables.hql pour créer des tables Hive cust_parquet (Parquet) et cust_orc (ORC) dans votre bucket Cloud Storage. 

      gcloud dataproc jobs submit hive \
    --cluster=KAFKA_CLUSTER \
    --region=REGION \
    -f ~/hivetables.hql

      Remarques :

      • Le composant Hive est préinstallé sur le cluster Dataproc Kafka. Consultez les versions 2.1.x pour obtenir la liste des versions du composant Hive incluses dans les images 2.1 récemment publiées.
      • KAFKA_CLUSTER : nom de votre cluster Kafka.
      • REGION : région dans laquelle se trouve votre cluster Kafka.

    Transférer par flux Kafka custdata vers des tables Hive

    1. Exécutez la commande suivante dans le terminal SSH sur le nœud maître de votre cluster Kafka pour installer la bibliothèque kafka-python. Un client Kafka est nécessaire pour diffuser les données des sujets Kafka vers Cloud Storage.
      pip install kafka-python

    2. Insérez votre BUCKET_NAME, puis copiez et collez le code PySpark suivant dans le terminal SSH sur le nœud maître de votre cluster Kafka, puis appuyez sur <return> pour créer un fichier streamdata.py.

      Le script s'abonne au sujet Kafka custdata, puis diffuse les données vers vos tables Hive dans Cloud Storage. Le format de sortie, qui peut être parquet ou ORC, est transmis au script en tant que paramètre.

      cat > streamdata.py <<EOF
#!/bin/python

import sys
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql import SparkSession
from kafka import KafkaConsumer

def getNameFn (data): return data.split(",")[0]
def getAgeFn  (data): return data.split(",")[1]
def getAmtFn  (data): return data.split(",")[2]

def main(cluster, outputfmt):
    spark = SparkSession.builder.appName("APP").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    Logger = spark._jvm.org.apache.log4j.Logger
    logger = Logger.getLogger(__name__)

    rows = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", cluster+"-w-0:9092").option("subscribe", "custdata") \
    .option("startingOffsets", "earliest")\
    .load()

    getNameUDF = udf(getNameFn, StringType())
    getAgeUDF  = udf(getAgeFn,  StringType())
    getAmtUDF  = udf(getAmtFn,  StringType())

    logger.warn("Params passed in are cluster name: " + cluster + "  output format(sink): " + outputfmt)

    query = rows.select (col("key").cast("string").alias("uuid"),\
        getNameUDF      (col("value").cast("string")).alias("custname"),\
        getAgeUDF       (col("value").cast("string")).alias("age"),\
        getAmtUDF       (col("value").cast("string")).alias("amount"))

    writer = query.writeStream.format(outputfmt)\
            .option("path","gs://BUCKET_NAME/tables/cust_"+outputfmt)\
            .option("checkpointLocation", "gs://BUCKET_NAME/chkpt/"+outputfmt+"wr") \
        .outputMode("append")\
        .start()

    writer.awaitTermination()

if __name__=="__main__":
    if len(sys.argv) < 2:
        print ("Invalid number of arguments passed ", len(sys.argv))
        print ("Usage: ", sys.argv[0], " cluster  format")
        print ("e.g.:  ", sys.argv[0], " <cluster_name>  orc")
        print ("e.g.:  ", sys.argv[0], " <cluster_name>  parquet")
    main(sys.argv[1], sys.argv[2])

EOF

    3. Dans le terminal SSH du nœud maître de votre cluster Kafka, exécutez spark-submit pour diffuser des données dans vos tables Hive dans Cloud Storage.

      1. Insérez le nom de votre KAFKA_CLUSTER et la sortie FORMAT, puis copiez et collez le code suivant dans le terminal SSH du nœud maître de votre cluster Kafka. Appuyez ensuite sur <return> pour exécuter le code et diffuser les données custdata Kafka au format Parquet dans vos tables Hive dans Cloud Storage. 

        spark-submit --packages \
org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.3,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 \
    --conf spark.history.fs.gs.outputstream.type=FLUSHABLE_COMPOSITE \
    --conf spark.driver.memory=4096m \
    --conf spark.executor.cores=2 \
    --conf spark.executor.instances=2 \
    --conf spark.executor.memory=6144m \
    streamdata.py KAFKA_CLUSTER FORMAT

        Remarques :

        • KAFKA_CLUSTER : insérez le nom de votre cluster Kafka.
        • FORMAT : spécifiez parquet ou orc comme format de sortie. Vous pouvez exécuter la commande successivement pour diffuser les deux formats dans les tables Hive. Par exemple, lors de la première invocation, spécifiez parquet pour diffuser le thème Kafka custdata dans la table Parquet Hive. Ensuite, lors de la deuxième invocation, spécifiez le format orc pour diffuser custdata dans la table ORC Hive.

    4. Une fois que la sortie standard s'arrête dans le terminal SSH, ce qui signifie que tous les custdata ont été diffusés, appuyez sur <control-c> dans le terminal SSH pour arrêter le processus.

    5. Répertoriez les tables Hive dans Cloud Storage.

      gcloud storage ls gs://BUCKET_NAME/tables/* --recursive

      Remarques :

      • BUCKET_NAME : insérez le nom du bucket Cloud Storage contenant vos tables Hive (voir Créer des tables Hive).

    Interroger les données diffusées

    1. Dans le terminal SSH du nœud maître de votre cluster Kafka, exécutez la commande hive suivante pour compter les messages custdata Kafka diffusés dans les tables Hive de Cloud Storage.

      hive -e "select count(1) from TABLE_NAME"

      Remarques :

      • TABLE_NAME : spécifiez cust_parquet ou cust_orc comme nom de table Hive.

      Extrait de résultat attendu :

    ...
Status: Running (Executing on YARN cluster with App id application_....)

----------------------------------------------------------------------------------------------
        VERTICES      MODE        STATUS  TOTAL  COMPLETED  RUNNING  PENDING  FAILED  KILLED  
----------------------------------------------------------------------------------------------
Map 1 .......... container     SUCCEEDED      1          1        0        0       0       0
Reducer 2 ...... container     SUCCEEDED      1          1        0        0       0       0
----------------------------------------------------------------------------------------------
VERTICES: 02/02  [==========================>>] 100%  ELAPSED TIME: 9.89 s     
----------------------------------------------------------------------------------------------
OK
10000
Time taken: 21.394 seconds, Fetched: 1 row(s)

    Effectuer un nettoyage

    Supprimer le projet

    1. In the Google Cloud console, go to the Manage resources page.

      Go to Manage resources

    2. In the project list, select the project that you want to delete, and then click Delete.
    3. In the dialog, type the project ID, and then click Shut down to delete the project.

    Delete resources

    • In the Google Cloud console, go to the Cloud Storage Buckets page.

      Go to Buckets

    • Click the checkbox for the bucket that you want to delete.
    • To delete the bucket, click Delete, and then follow the instructions.
    • Supprimez votre cluster Kafka : 
      gcloud dataproc clusters delete KAFKA_CLUSTER \
    --region=${REGION}