基于Apache Hudi在Google云创建数据湖平台的思路详解

发布时间：2023-02-20 11:01:24 所属栏目：Apache 来源：互联网

导读：自从计算机出现以来，我们一直在尝试寻找计算机存储一些信息的方法，存储在计算机上的信息（也称为数据）有多种形式，数据变得如此重要，以至于信息现在已成为触手可及的商品。多年来数据以多种方式存储在计算机中，包括数据库、blob存储和其他方法，为了进

　　自从计算机出现以来，我们一直在尝试寻找计算机存储一些信息的方法，存储在计算机上的信息（也称为数据）有多种形式，数据变得如此重要，以至于信息现在已成为触手可及的商品。多年来数据以多种方式存储在计算机中，包括数据库、blob存储和其他方法，为了进行有效的业务分析，必须对现代应用程序创建的数据进行处理和分析，并且产生的数据量非常巨大！有效地存储数PB数据并拥有必要的工具来查询它以便使用它至关重要，只有这样对该数据的分析才能产生有意义的结果。

　　Debezium
　　MySQL
　　Apache Kafka
　　Apache Hudi
　　Apache Spark
　　我们将要构建的数据湖架构如下：

　　第一步是使用 Debezium 读取关系数据库中发生的所有更改，并将所有更改推送到 Kafka 集群。

　　Debezium 是一个用于变更数据捕获的开源分布式平台，Debezium 可以指向任何关系数据库，并且它可以开始实时捕获任何数据更改，它非常快速且实用，由红帽维护。

　　首先，我们将使用 docker-compose 在我们的机器上设置 Debezium、MySQL 和 Kafka，您也可以使用这些的独立安装，我们将使用 Debezium 提供给我们的 mysql 镜像，因为其中已经包含数据，在任何生产环境中都可以使用适当的 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 可以设置为 1.8。此外请确保设置 MYSQL_ROOT_PASS、MYSQL_USER 和 MYSQL_PASSWORD。

　　在我们继续之前，我们将查看 debezium 镜像提供给我们的数据库 inventory 的结构，进入数据库的命令行：

　　1
　　docker-compose -f docker-compose-avro-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'
　　在 shell 内部，我们可以使用 show tables 命令。输出应该是这样的：

　　我们可以通过 select * from customers 命令来查看客户表的内容。输出应该是这样的：

　　现在在创建容器后，我们将能够为 Kafka Connect 激活 Debezium 源连接器，我们将使用的数据格式是 Avro 数据格式，Avro 是在 Apache 的 Hadoop 项目中开发的面向行的远程过程调用和数据序列化框架。它使用 JSON 来定义数据类型和协议，并以紧凑的二进制格式序列化数据。

　　让我们用我们的 Debezium 连接器的配置创建另一个文件。

　　{
　　    "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"
　　    }
　　}
　　正如我们所看到的，我们已经在其中配置了数据库的详细信息以及要从中读取更改的数据库，确保将 MYSQL_USER 和 MYSQL_PASSWORD 的值更改为您之前配置的值，现在我们将运行一个命令在 Kafka Connect 中注册它，命令如下：

　　1
　　curl -i -X POST -H "Accept:application/json" -H "Content-type:application/json" http://localhost:8083/connectors/ -d @register-mysql.json
　　现在，Debezium 应该能够从 Kafka 读取数据库更改。
　　下一步涉及使用 Spark 和 Hudi 从 Kafka 读取数据，并将它们以 Hudi 文件格式放入 Google Cloud Storage Bucket。在我们开始使用它们之前，让我们了解一下 Hudi 和 Spark 是什么。

　　Apache Hudi 是一个开源数据管理框架，用于简化增量数据处理和数据管道开发。该框架更有效地管理数据生命周期等业务需求并提高数据质量。 Hudi 使您能够在基于云的数据湖上管理记录级别的数据，以简化更改数据捕获 (CDC) 和流式数据摄取，并帮助处理需要记录级别更新和删除的数据隐私用例。 Hudi 管理的数据集使用开放存储格式存储在云存储桶中，而与 Presto、Apache Hive 和/或 Apache Spark 的集成使用熟悉的工具提供近乎实时的更新数据访问

（编辑：甘南站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!

1/2

尾页