Apache RocketMQ + Hudi 快速構建 Lakehouse

本文目錄

• 背景知識

• 大資料時代的構架演進

• RocketMQ Connector&Stream

• Apache Hudi

• 構建Lakehouse實操

本文標題包含三個關鍵詞：Lakehouse、RocketMQ、Hudi。我們先從整體Lakehouse架構入手，隨後逐步分析架構產生的原因、架構元件特點以及構建Lakehouse架構的實操部分。

01

背 景 知 識

1 、Lakehouse架構

Lakehouse最初由Databrick提出，並對Lakehouse架構特徵有如下要求：

（1）事務支援

企業內部許多資料管道通常會併發讀寫資料。對ACID事務的支援確保了多方併發讀寫資料時的一致性問題；

（2）Schema enforcement and governance

Lakehouse應該有一種方式可以支援模式執行和演進、支援DW schema的正規化（如星星或雪花模型），能夠對資料完整性進行推理，並且具有健壯的治理和審計機制；

（3）開放性

使用的儲存格式是開放式和標準化的（如parquet），並且為各類工具和引擎，包括機器學習和Python/R庫，提供API，以便它們可以直接有效地訪問資料；

（4）BI支援

Lakehouse可以直接在源資料上使用BI工具。這樣可以提高資料新鮮度、減少延遲，並且降低了在資料池和資料倉庫中操作兩個資料副本的成本；

（5）儲存與計算分離

在實踐中，這意味著儲存和計算使用單獨的叢集，因此這些系統能夠擴充套件到支援更大的使用者併發和資料量。一些現代數倉也具有此屬性；

（6）支援從非結構化資料到結構化資料的多種資料型別

Lakehouse可用於儲存、優化、分析和訪問許多資料應用所需的包括image、video、audio、text以及半結構化資料；

（7）支援各種工作負載

包括資料科學、機器學習以及SQL和分析。可能需要多種工具來支援這些工作負載，但它們底層都依賴同一資料儲存庫；

（8）端到端流

實時報表是許多企業中的標準應用。對流的支援消除了需要構建單獨系統來專門用於服務實時資料應用的需求。

從上述對Lakehouse架構的特點描述我們可以看出，針對單一功能，我們可以利用某些開源產品組合構建出一套解決方案。但對於全部功能的支援，目前好像沒有一個通用的解決方案。接下來，我們先了解大資料時代主流的資料處理架構是怎樣的。

02

大資料時代的架構演進

為什麼會產生這麼多開源產品呢？首先在大資料時代資料量越來越大，而且每個業務的需求也各不相同，因此就產生出各種型別的產品供架構師選擇，用於支援各類場景。然而眾多的品類產品也給架構師們帶來一些困擾，比如選型困難、試錯成本高、學習成本高、架構複雜等等。

03

RocketMQ Connector & Stream

 nohup sh bin/mqnamesrv &
 nohup sh bin/mqbroker -n localhost:9876 &

kubectl apply -f example/rocketmq_cluster.yaml

04

Apache Hudi

05

構建 Lakehouse 實操

rocketmq-connect-hudi version：0.0.1-SNAPSHOT

/data/lakehouse/rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime/target/distribution/conf/connect.conf 中connector-plugin 路徑

cd rocketmq-externals/rocketmq-connect-hudi
mvn clean install -DskipTest -U

cd /data/lakehouse/rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime
  sh ./run_worker.sh    ##  Worker可以啟動多個

curl http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-hudi-sink-connector-name} ?config='{"connector-class":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"topicc","tablePath":"file:///tmp/hudi_connector_test","tableName":"hudi_connector_test_table","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","source-rocketmq":"127.0.0.1:9876","src-cluster":"DefaultCluster","refresh-interval":"10000","schemaPath":"/data/lakehouse/config/user.avsc"\}’
  啟動成功會列印如下日誌：
2021-09-06 16:23:14 INFO pool-2-thread-1 - Open HoodieJavaWriteClient successfully

RocketMQ connector配置表

(2) Hudi任務需要配置Hudi表路徑tablePath和表名稱tableName，以及Hudi使用的Schema檔案。

Hudi任務配置表

點選閱讀原文即可檢視Lakehouse構建實操影片

附錄： 在本地Mac系統構建Lakehouse demo

涉及到的元件： ro cketmq、rocketmq-connector-runtime、rocketmq-connect-hudi、hudi、hdfs、avro、spark-shell

0、啟動hdfs
下載hadoop包

https://www.apache.org/dyn/closer.cgi/hadoop/common/hadoop-2.10.1/hadoop-2.10.1.tar.gz

cd /Users/osgoo/Documents/hadoop-2.10.1
vi core-site.xml
<configuration>
<property>
 <name>fs.defaultFS</name>
 <!-- 可以通過命令hostname 檢視主機名字  這裡的主機名字是hadoop1-->
 <value>hdfs://localhost:9000</value>
</property>
<!--覆蓋掉core-default.xml中的預設配置-->
</configuration>


vi hdfs-site.xml
<configuration>
<property>
        <name>dfs.replication</name>
        <value>1</value>
  </property>
</configuration>


./bin/hdfs namenode -format
./sbin/start-dfs.sh 
jps 看下namenode,datanode
lsof -i:9000
./bin/hdfs dfs -mkdir -p /Users/osgoo/Downloads


1、啟動rocketmq叢集，建立rocketmq-connector內建topic
QickStart：https://rocketmq.apache.org/docs/quick-start/
sh mqadmin updatetopic -t connector-cluster-topic -n localhost:9876 -c DefaultCluster
sh mqadmin updatetopic -t connector-config-topic -n localhost:9876 -c DefaultCluster
sh mqadmin updatetopic -t connector-offset-topic -n localhost:9876 -c DefaultCluster
sh mqadmin updatetopic -t connector-position-topic -n localhost:9876 -c DefaultCluster


2、建立資料入湖的源端topic，testhudi1
sh mqadmin updatetopic -t testhudi1 -n localhost:9876 -c DefaultCluster


3、編譯rocketmq-connect-hudi-0.0.1-SNAPSHOT-jar-with-dependencies.jar
cd rocketmq-connect-hudi
mvn clean install -DskipTest -U


4、啟動rocketmq-connector runtime
配置connect.conf
--------------
workerId=DEFAULT_WORKER_1
storePathRootDir=/Users/osgoo/Downloads/storeRoot


## Http port for user to access REST API
httpPort=8082


# Rocketmq namesrvAddr
namesrvAddr=localhost:9876


# Source or sink connector jar file dir,The default value is rocketmq-connect-sample
pluginPaths=/Users/osgoo/Downloads/connector-plugins
---------------
拷貝 rocketmq-hudi-connector.jar 到 pluginPaths=/Users/osgoo/Downloads/connector-plugins


sh run_worker.sh


5、配置入湖config
curl http://localhost:8082/connectors/rocketmq-connect-hudi?config='\{"connector-class":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"testhudi1","tablePath":"hdfs://localhost:9000/Users/osgoo/Documents/base-path7","tableName":"t7","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","source-rocketmq":"127.0.0.1:9876","source-cluster":"DefaultCluster","refresh-interval":"10000","schemaPath":"/Users/osgoo/Downloads/user.avsc"\}'


6、傳送訊息到testhudi1




7、## 利用spark讀取


cd /Users/osgoo/Downloads/spark-3.1.2-bin-hadoop3.2/bin


./spark-shell \
  --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.9.0,org.apache.spark:spark-avro_2.12:3.0.1 \
  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'


import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._


val tableName = "t7"
val basePath = "hdfs://localhost:9000/Users/osgoo/Documents/base-path7"


val tripsSnapshotDF = spark.
  read.
  format("hudi").
  load(basePath + "/*")
tripsSnapshotDF.createOrReplaceTempView("hudi_trips_snapshot")


spark.sql("select * from hudi_trips_snapshot").show()

END