将 Kafka 主题流式传输到 Hive

教程步骤

请执行以下步骤创建 Managed Service for Apache Spark Kafka 集群，以 Parquet 或 ORC 格式将 Kafka 主题读入 Cloud Storage。

将 Kafka 安装脚本复制到 Cloud Storage

kafka.sh 初始化操作脚本会在 Managed Service for Apache Spark 集群上安装 Kafka。

1. 浏览代码。

   展开 收起

   #!/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() {
     if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} >= 3.0" | bc -l) == 1 ]]; then
       retry_apt_command "apt-get update && apt-get install -y gnupg"
       export GNUPGHOME="$(mktemp -d)"
       trap 'rm -rf "${GNUPGHOME}"' EXIT
       gpg --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C
       mkdir -p /etc/apt/trusted.gpg.d
       gpg --export B7B3B788A8D3785C > /etc/apt/trusted.gpg.d/mysql-repo.gpg
     else
       retry_apt_command "apt-get install -y gnupg2 && \
         apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
     fi
   }
   
   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