自從計(jì)算機(jī)出現(xiàn)以來(lái)，我們一直在嘗試尋找計(jì)算機(jī)存儲(chǔ)一些信息的方法，存儲(chǔ)在計(jì)算機(jī)上的信息（也稱為數(shù)據(jù)）有多種形式，數(shù)據(jù)變得如此重要，以至于信息現(xiàn)在已成為觸手可及的商品。多年來(lái)數(shù)據(jù)以多種方式存儲(chǔ)在計(jì)算機(jī)中，包括數(shù)據(jù)庫(kù)、blob存儲(chǔ)和其他方法，為了進(jìn)行有效的業(yè)務(wù)分析，必須對(duì)現(xiàn)代應(yīng)用程序創(chuàng)建的數(shù)據(jù)進(jìn)行處理和分析，并且產(chǎn)生的數(shù)據(jù)量非常巨大！有效地存儲(chǔ)數(shù)PB數(shù)據(jù)并擁有必要的工具來(lái)查詢它以便使用它至關(guān)重要，只有這樣對(duì)該數(shù)據(jù)的分析才能產(chǎn)生有意義的結(jié)果。
大數(shù)據(jù)是一門處理分析方法、有條不紊地從中提取信息或以其他方式處理對(duì)于典型數(shù)據(jù)處理應(yīng)用程序軟件而言過(guò)于龐大或復(fù)雜的數(shù)據(jù)量的方法的學(xué)科。為了處理現(xiàn)代應(yīng)用程序產(chǎn)生的數(shù)據(jù)，大數(shù)據(jù)的應(yīng)用是非常必要的，考慮到這一點(diǎn)，本博客旨在提供一個(gè)關(guān)于如何創(chuàng)建數(shù)據(jù)湖的小教程，該數(shù)據(jù)湖從應(yīng)用程序的數(shù)據(jù)庫(kù)中讀取任何更改并將其寫入數(shù)據(jù)湖中的相關(guān)位置，我們將為此使用的工具如下：

• Debezium
• MySQL
• Apache Kafka
• Apache Hudi
• Apache Spark

我們將要構(gòu)建的數(shù)據(jù)湖架構(gòu)如下：

第一步是使用 Debezium 讀取關(guān)系數(shù)據(jù)庫(kù)中發(fā)生的所有更改，并將所有更改推送到 Kafka 集群。

Debezium 是一個(gè)用于變更數(shù)據(jù)捕獲的開(kāi)源分布式平臺(tái)，Debezium 可以指向任何關(guān)系數(shù)據(jù)庫(kù)，并且它可以開(kāi)始實(shí)時(shí)捕獲任何數(shù)據(jù)更改，它非常快速且實(shí)用，由紅帽維護(hù)。

首先，我們將使用 docker-compose 在我們的機(jī)器上設(shè)置 Debezium、MySQL 和 Kafka，您也可以使用這些的獨(dú)立安裝，我們將使用 Debezium 提供給我們的 mysql 鏡像，因?yàn)槠渲幸呀?jīng)包含數(shù)據(jù)，在任何生產(chǎn)環(huán)境中都可以使用適當(dāng)?shù)?Kafka、MySQL 和 Debezium 集群，docker compose 文件如下：

version: '2'
services:
  zookeeper:
    image: debezium/zookeeper:${DEBEZIUM_VERSION}
    ports:
     - 2181:2181
     - 2888:2888
     - 3888:3888
  kafka:
    image: debezium/kafka:${DEBEZIUM_VERSION}
    ports:
     - 9092:9092
    links:
     - zookeeper
    environment:
     - ZOOKEEPER_CONNECT=zookeeper:2181
  mysql:
    image: debezium/example-mysql:${DEBEZIUM_VERSION}
    ports:
     - 3307:3306
    environment:
     - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASS}
     - MYSQL_USER=${MYSQL_USER}
     - MYSQL_PASSWORD=${MYSQL_USER_PASS}
  schema-registry:
    image: confluentinc/cp-schema-registry
    ports:
     - 8181:8181
     - 8081:8081
    environment:
     - SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092
     - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL=zookeeper:2181
     - SCHEMA_REGISTRY_HOST_NAME=schema-registry
     - SCHEMA_REGISTRY_LISTENERS=http://schema-registry:8081
    links:
     - zookeeper
  connect:
    image: debezium/connect:${DEBEZIUM_VERSION}
    ports:
     - 8083:8083
    links:
     - kafka
     - mysql
     - schema-registry
    environment:
     - BOOTSTRAP_SERVERS=kafka:9092
     - GROUP_ID=1
     - CONFIG_STORAGE_TOPIC=my_connect_configs
     - OFFSET_STORAGE_TOPIC=my_connect_offsets
     - STATUS_STORAGE_TOPIC=my_connect_statuses
     - KEY_CONVERTER=io.confluent.connect.avro.AvroConverter
     - VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter
     - INTERNAL_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
     - INTERNAL_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
     - CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL=http://schema-registry:8081
     - CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL=http://schema-registry:8081

DEBEZIUM_VERSION 可以設(shè)置為 1.8。 此外請(qǐng)確保設(shè)置 MYSQL_ROOT_PASS、MYSQL_USER 和 MYSQL_PASSWORD。

在我們繼續(xù)之前，我們將查看 debezium 鏡像提供給我們的數(shù)據(jù)庫(kù) inventory 的結(jié)構(gòu)，進(jìn)入數(shù)據(jù)庫(kù)的命令行：

docker-compose -f docker-compose-avro-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

在 shell 內(nèi)部，我們可以使用 show tables 命令。 輸出應(yīng)該是這樣的：

我們可以通過(guò) select * from customers 命令來(lái)查看客戶表的內(nèi)容。 輸出應(yīng)該是這樣的：

現(xiàn)在在創(chuàng)建容器后，我們將能夠?yàn)?Kafka Connect 激活 Debezium 源連接器，我們將使用的數(shù)據(jù)格式是 Avro 數(shù)據(jù)格式，Avro 是在 Apache 的 Hadoop 項(xiàng)目中開(kāi)發(fā)的面向行的遠(yuǎn)程過(guò)程調(diào)用和數(shù)據(jù)序列化框架。它使用 JSON 來(lái)定義數(shù)據(jù)類型和協(xié)議，并以緊湊的二進(jìn)制格式序列化數(shù)據(jù)。

讓我們用我們的 Debezium 連接器的配置創(chuàng)建另一個(gè)文件。

{
    "name": "inventory-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "mysql",
        "database.port": "3306",
        "database.user": "MYSQL_USER",
        "database.password": "MYSQL_PASSWORD",
        "database.server.id": "184054",
        "database.server.name": "dbserver1",
        "database.include.list": "inventory",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://schema-registry:8081",
        "value.converter.schema.registry.url": "http://schema-registry:8081"
    }
}

正如我們所看到的，我們已經(jīng)在其中配置了數(shù)據(jù)庫(kù)的詳細(xì)信息以及要從中讀取更改的數(shù)據(jù)庫(kù)，確保將 MYSQL_USER 和 MYSQL_PASSWORD 的值更改為您之前配置的值，現(xiàn)在我們將運(yùn)行一個(gè)命令在 Kafka Connect 中注冊(cè)它，命令如下：

curl -i -X POST -H "Accept:application/json" -H "Content-type:application/json" http://localhost:8083/connectors/ -d @register-mysql.json

現(xiàn)在，Debezium 應(yīng)該能夠從 Kafka 讀取數(shù)據(jù)庫(kù)更改。
下一步涉及使用 Spark 和 Hudi 從 Kafka 讀取數(shù)據(jù)，并將它們以 Hudi 文件格式放入 Google Cloud Storage Bucket。 在我們開(kāi)始使用它們之前，讓我們了解一下 Hudi 和 Spark 是什么。

Apache Hudi 是一個(gè)開(kāi)源數(shù)據(jù)管理框架，用于簡(jiǎn)化增量數(shù)據(jù)處理和數(shù)據(jù)管道開(kāi)發(fā)。 該框架更有效地管理數(shù)據(jù)生命周期等業(yè)務(wù)需求并提高數(shù)據(jù)質(zhì)量。 Hudi 使您能夠在基于云的數(shù)據(jù)湖上管理記錄級(jí)別的數(shù)據(jù)，以簡(jiǎn)化更改數(shù)據(jù)捕獲 (CDC) 和流式數(shù)據(jù)攝取，并幫助處理需要記錄級(jí)別更新和刪除的數(shù)據(jù)隱私用例。 Hudi 管理的數(shù)據(jù)集使用開(kāi)放存儲(chǔ)格式存儲(chǔ)在云存儲(chǔ)桶中，而與 Presto、Apache Hive 和/或 Apache Spark 的集成使用熟悉的工具提供近乎實(shí)時(shí)的更新數(shù)據(jù)訪問(wèn)

Apache Spark 是用于大規(guī)模數(shù)據(jù)處理的開(kāi)源統(tǒng)一分析引擎。 Spark 為具有隱式數(shù)據(jù)并行性和容錯(cuò)性的集群編程提供了一個(gè)接口。 Spark 代碼庫(kù)最初是在加州大學(xué)伯克利分校的 AMPLab 開(kāi)發(fā)的，后來(lái)被捐贈(zèng)給了 Apache 軟件基金會(huì)，該基金會(huì)一直在維護(hù)它。

現(xiàn)在，由于我們正在 Google Cloud 上構(gòu)建解決方案，因此最好的方法是使用 Google Cloud Dataproc。 Google Cloud Dataproc 是一種托管服務(wù)，用于處理大型數(shù)據(jù)集，例如大數(shù)據(jù)計(jì)劃中使用的數(shù)據(jù)集。 Dataproc 是 Google 的公共云產(chǎn)品 Google Cloud Platform 的一部分。 Dataproc 幫助用戶處理、轉(zhuǎn)換和理解大量數(shù)據(jù)。

在 Google Dataproc 實(shí)例中，預(yù)裝了 Spark 和所有必需的庫(kù)。 創(chuàng)建實(shí)例后，我們可以在其中運(yùn)行以下 Spark 作業(yè)來(lái)完成我們的管道：

spark-submit \
  --packages org.apache.hudi:hudi-spark3.1.2-bundle_2.12:0.10.1,org.apache.spark:spark-avro_2.12:3.1.2 \
  --master yarn --deploy-mode client \
  --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /usr/lib/hadoop/hudi-packages/hudi-utilities-bundle_2.12-0.10.1.jar \
  --table-type COPY_ON_WRITE --op UPSERT \
  --target-base-path gs://your-data-lake-bucket/hudi/customers \
  --target-table hudi_customers --continuous \
  --min-sync-interval-seconds 60 \
  --source-class org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource \
  --source-ordering-field _event_origin_ts_ms \
  --hoodie-conf schema.registry.url=http://localhost:8081 \
  --hoodie-conf hoodie.deltastreamer.schemaprovider.registry.url=http://localhost:8081/subjects/dbserver1.inventory.customers-value/versions/latest \
  --hoodie-conf hoodie.deltastreamer.source.kafka.topic=dbserver1.inventory.customers \
  --hoodie-conf bootstrap.servers=localhost:9092 \
  --hoodie-conf auto.offset.reset=earliest \
  --hoodie-conf hoodie.datasource.write.recordkey.field=id \
  --hoodie-conf hoodie.datasource.write.partitionpath.field=id \

這將運(yùn)行一個(gè) spark 作業(yè)，該作業(yè)從我們之前推送到的 Kafka 中獲取數(shù)據(jù)并將其寫入 Google Cloud Storage Bucket。 我們必須指定 Kafka 主題、Schema Registry URL 和其他相關(guān)配置。

結(jié)論

可以通過(guò)多種方式構(gòu)建數(shù)據(jù)湖。 我試圖展示如何使用 Debezium、Kafka、Hudi、Spark 和 Google Cloud 構(gòu)建數(shù)據(jù)湖。 使用這樣的設(shè)置，可以輕松擴(kuò)展管道以管理大量數(shù)據(jù)工作負(fù)載！ 有關(guān)每種技術(shù)的更多詳細(xì)信息，可以訪問(wèn)文檔。 可以自定義 Spark 作業(yè)以獲得更細(xì)粒度的控制。 這里顯示的 Hudi 也可以與 Presto、Hive 或 Trino 集成。 定制的數(shù)量是無(wú)窮無(wú)盡的。 本文提供了有關(guān)如何使用上述工具構(gòu)建基本數(shù)據(jù)管道的基本介紹！

到此這篇關(guān)于基于Apache Hudi在Google云構(gòu)建數(shù)據(jù)湖平臺(tái)的文章就介紹到這了,更多相關(guān)Apache Hudi構(gòu)建數(shù)據(jù)湖內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！