DataLakeAnalysis: 使用DataX同步Kafka数据到OSS进行分析

简介:

平常业务开发中我们经常有流式数据保存在Kafka里面,这部分数据很多场景也是需要分析的,今天给大家介绍下如果使用DataX把数据从Kafka同步到OSS,保存成对分析友好的Parquet格式,然后利用DLA进行分析的全流程。
为了后续叙述的方便,我们假设Kafka里面保存的是订单的数据,它包含如下字段:

  • id, int类型
  • name, string类型
  • gmt_create, bigint类型, 时间戳字段
  • map_col: MAP 类型
  • array_col: ARRAY 类型
  • struct_col: STRUCT 类型

我们最终希望把Kafka上的这个数据最终保存到OSS上,并且映射到DLA里面的一个分区表,表结构如下:

CREATE EXTERNAL TABLE orders_p (
    id int,
    name string,
    gmt_create timestamp,
    map_col MAP<string, string>,
    array_col ARRAY<string>,
    struct_col STRUCT<id:bigint,name:string>
)
PARTITIONED BY (dt string)
STORED AS PARQUET 
LOCATION 'oss://test-bucket/datasets/oss_demo/orders_p/';

注意我们最终表结构里面有一个分区字段 dt, 因为分析场景下数据量都很大,进行分区才能提高分析的效率。而这个分区字段在原始数据里面是没有直接对应的。
因为从DataX过来的数据无法自动根据目录分区,因此我们建议从Kafka过来的数据放到中间表 orders 里面去, 在数据进入orders之后我们再跟一个任务做一个“数据拆分”的工作,把数据拆分到具体的分区里面去。总结一下:要把Kafka的数据正确的落到DLA里面的 orders_p 需要经过如下步骤:

  • kafka -> oss: DataX定时把数据同步到中间表: orders。
  • orders -> orders_p: DLA任务定时把数据从中间表orders同步到orders_p

这两个任务串行执行,第二个任务依赖第一个任务,每5分钟调度一次。这样就可以Kafka里面的数据以5分钟延时的粒度不断地写入到OSS里面去,然后使用DLA进行高效的分析。

kafka -> oss: DataX定时把数据同步到中间表: orders

因为Kafka上的数据量很大,在DLA中一般会进行分区处理以获得更好的分析性能,但是DataX目前还无法支持直接把数据写入到分区表,因此我们要搞一个中间表: orders 过度一下,它的表结构跟最终表orders_p几乎一样,只是没有分区

CREATE EXTERNAL TABLE orders (
    id int,
    name string,
    gmt_create timestamp,
    map_col MAP<string, string>,
    array_col ARRAY<string>,
    struct_col STRUCT<id:bigint,name:string>
)
STORED AS PARQUET 
LOCATION 'oss://test-bucket/datasets/oss_demo/orders/';

那么我们第一步要做的事情就是要通过DataX把数据写到这个 orders 表对应的LOCATION: oss://test-bucket/datasets/oss_demo/orders/。
整个DataX的任务的JSON配置蛮复杂的,我们直接贴在这里:

{
    "job": {
        "setting": {
            "speed": {
                "channel": 3
            },
            "errorLimit": {
                "record": 5
            }
        },
        "content": [
            {
                "reader": {
                    "name": "kafkareader",
                    "parameter": {
                        "server": "127.0.0.2:9093",
                        "column": [
                            "id",
                            "name",
                            "gmt_create",
                            "map_col",
                            "array_col",
                            "struct_col"
                        ],
                        "kafkaConfig": {
                            "group.id": "demo_test",
                            "java.security.auth.login.config": "/the-path/kafka/kafka_client_jaas.conf",
                            "ssl.truststore.location": "/the-path/kafka.client.truststore.jks",
                            "ssl.truststore.password": "KafkaOnsClient",
                            "security.protocol": "SASL_SSL",
                            "sasl.mechanism": "PLAIN",
                            "ssl.endpoint.identification.algorithm": ""
                        },
                        "topic": "yucha",
                        "waitTime": "10",
                        "partition_": "0",
                        "keyType": "ByteArray",
                        "valueType": "ByteArray",
                        "seekToBeginning_": "true",
                        "seekToLast_": "true",
                        "beginDateTime": "20190501010000",
                        "endDateTime":   "20190501010500"
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "oss://test-bucket",
                        "fileType": "parquet",
                        "path": "/datasets/oss_demo/kpt",
                        "fileName": "test",
                        "writeMode": "truncate",
                        "compress":"SNAPPY",
                        "encoding":"UTF-8",
                        "hadoopConfig": {
                            "fs.oss.accessKeyId": "the-access-id",
                            "fs.oss.accessKeySecret": "the-access-key",
                            "fs.oss.endpoint": "oss-cn-hangzhou.aliyuncs.com"
                        },
                        "parquetSchema": "message test {\n    required int64 id;\n    optional binary name (UTF8);\n    optional int64 gmt_create;\n    required group map_col (MAP) {\n        repeated group key_value {\n            required binary key (UTF8);\n            required binary value (UTF8);\n        }\n    }\n    required group array_col (LIST) {\n        repeated group list {\n            required binary element (UTF8);\n        }\n    }\n    required group struct_col {\n        required int64 id;\n        required binary name (UTF8);\n    }    \n}",
                        "dataxParquetMode": "fields"
                    }
                }
            }
        ]
    }
}

这个配置分为两段: kafkareader 和 hdfswriter, 分别负责读写数据。我们分别详细介绍一下。
kafkareader 里面大多数参数都比较好理解,比较重要的参数是 beginDateTime, endDateTime, 指定这个任务要消费的kafka数据的范围,比如我们任务每5分钟跑一次,那么这里指定的可能就是当前时间往前推5分钟的时间范围,比如我们示例代码里面的是 20190501010000 到 20190501010500, 时间精确到秒。关于KafkaReader更详细的信息可以参考KafkaReader文档。
hdfswriter, 这里我们使用hdfswriter来写oss数据是因为OSS实现了Hadoop File System的接口,我们可以通过HDFS Writer来向OSS导数据,因为倒过来的数据后面要通过DLA来分析,推荐使用Parquet这种列存格式来保存,目前HDFS Writer支持PARQUET的绝大部分类型,包括基本类型以及复杂类型如array, map, struct, 要以Parquet格式同步数据,我们首先要描述一下这个Parquet的格式, 我们示例数据对应的Parquet的Schema如下:

message test {
    required int64 id;
    optional binary name (UTF8);
    optional int64 gmt_create;
    required group map_col (MAP) {
        repeated group key_value {
            required binary key (UTF8);
            required binary value (UTF8);
        }
    }
    required group array_col (LIST) {
        repeated group list {
            required binary element (UTF8);
        }
    }
    required group struct_col {
        required int64 id;
        required binary name (UTF8);
    }    
}

上面DataX任务描述文件里面的parquetSchema字段里面的内容就是上面这段,只不过缩成了一行以保证整个DataX描述文件符合JSON格式。 关于Parquet Schema更多的信息可以查看Parquet Logical Type Definitions。
另外一个注意的配置点是 writeMode, 在我们的这个方案里面,我们推荐使用 truncate, 因为这个任务是每5分钟调度一次,下一次执行的时候需要把前一次执行的数据清空掉(truncate)。

orders -> orders_p: DLA任务定时把数据从中间表orders同步到orders_p

拆分的SQL每个具体的业务会不一样,我们这个示例里面比较简单,主要干了两件事:
把原始的 bigint 类型的 gmt_create转成了timestamp类型。
从gmt_create里面生成新的dt字段。

INSERT INTO orders_p
SELECT 
id, 
name, 
from_unixtime(gmt_create),  -- bigint -> timestamp
map_col, array_col, struct_col, 
cast(date(from_unixtime(gmt_create)) as string) -- 添加分区字段 
FROM orders

这两个任务的串行操作可以通过任务调度服务比如阿里云上DataWorks来进行串联,在 kafka -> oss 的任务完成后,运行这个 数据拆分 的任务。

总结

这篇文章介绍了如何把Kafka里面的数据实时地流入OSS,利用DLA进行高效的数据分析。借助于DataX对于Parquet复杂类型的支持,我们已经可以帮助用户把各种复杂数据搬进OSS,希望对有类似场景的客户有所帮助。

欢迎关注数据湖技术社区

数据湖开发者社区由 阿里云开发者社区 与 阿里云Data Lake Analytics团队 共同发起,致力于推广数据湖相关技术,包括hudi、delta、spark、presto、oss、元数据、存储加速、格式发现等,学习如何构建数据湖分析系统,打造适合业务的数据架构。

x

目录
相关文章
|
5月前
|
机器学习/深度学习 人工智能 专有云
人工智能平台PAI使用问题之怎么将DLC的数据写入到另一个阿里云主账号的OSS中
阿里云人工智能平台PAI是一个功能强大、易于使用的AI开发平台,旨在降低AI开发门槛,加速创新,助力企业和开发者高效构建、部署和管理人工智能应用。其中包含了一系列相互协同的产品与服务,共同构成一个完整的人工智能开发与应用生态系统。以下是对PAI产品使用合集的概述,涵盖数据处理、模型开发、训练加速、模型部署及管理等多个环节。
|
4月前
|
存储 安全 大数据
对象存储的意义:探索数据新纪元的关键基石
在信息爆炸时代,数据成为核心资产,而高效安全的数据存储至关重要。对象存储作为一种新兴技术,起源于20世纪90年代,旨在解决传统文件系统的局限性。随着云计算和大数据技术的发展,它已成为关键技术之一。对象存储具备高可扩展性、高可靠性、低成本、易于管理和多协议支持等优点。它支撑大数据发展、推动云计算繁荣、助力企业数字化转型并保障数据安全。未来,对象存储将进一步提升性能,实现智能化管理,并与边缘计算融合,获得政策支持,成为数据新时代的关键基石。
178 3
|
5月前
|
消息中间件 算法 Kafka
面试题Kafka问题之Kafka的副本消息同步如何解决
面试题Kafka问题之Kafka的副本消息同步如何解决
88 4
|
4月前
|
消息中间件 Java Kafka
Kafka生产者同步和异步的JavaAPI代码演示
Kafka生产者同步和异步的JavaAPI代码演示
46 0
|
5月前
|
消息中间件 SQL 分布式计算
DataWorks产品使用合集之如何离线增量同步Kafka数据,并指定时间范围进行同步
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
5月前
|
DataWorks 安全 定位技术
DataWorks产品使用合集之如何同步OSS中的Parquet数据,并解析里面的数组成多个字段
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
1月前
|
分布式计算 Java 开发工具
阿里云MaxCompute-XGBoost on Spark 极限梯度提升算法的分布式训练与模型持久化oss的实现与代码浅析
本文介绍了XGBoost在MaxCompute+OSS架构下模型持久化遇到的问题及其解决方案。首先简要介绍了XGBoost的特点和应用场景,随后详细描述了客户在将XGBoost on Spark任务从HDFS迁移到OSS时遇到的异常情况。通过分析异常堆栈和源代码,发现使用的`nativeBooster.saveModel`方法不支持OSS路径,而使用`write.overwrite().save`方法则能成功保存模型。最后提供了完整的Scala代码示例、Maven配置和提交命令,帮助用户顺利迁移模型存储路径。
|
4月前
|
存储 机器学习/深度学习 弹性计算
阿里云EMR数据湖文件系统问题之OSS-HDFS全托管服务的问题如何解决
阿里云EMR数据湖文件系统问题之OSS-HDFS全托管服务的问题如何解决
|
5月前
|
消息中间件 分布式计算 DataWorks
DataWorks产品使用合集之如何使用Python和阿里云SDK读取OSS中的文件
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
5月前
|
存储 运维 安全
阿里云OSS的优势
【7月更文挑战第19天】阿里云OSS的优势
211 2