Google云平台构建数据ETL任务的最佳实践

在数据处理中，我们经常需要构建ETL的任务，对数据进行加载，转换处理后再写入到数据存储中。Google的云平台提供了多种方案来构建ETL任务，我也研究了一下这些方案，比较方案之间的优缺点，从而找到一个最适合我业务场景的方案。

假设我们的业务场景需要定期从Kafka中获取数据，经过一些数据清洗，数据关联，数据Enrich操作之后，把数据写入到Bigquery数据仓库，从而方便以后生成统计分析报表。

Google云平台提供了几个方案来完成这个任务：

1. Datafusion，通过在UI界面设计ETL pipeline，然后把Pipeline转换为Spark应用，部署在Dataproc上运行。

2. 编写Spark应用代码，然后在Dataproc上运行或者在K8S集群上通过Spark operator来调度执行。

3. 编写Apache Beam代码，通过Dataflow runner在VM上执行任务。

方案一的优点是基本不需要编写代码，在图形界面上即可完成Pipeline的设计。缺点是如果有一些额外的需求可能不太方便实现，另外最主要是太贵。Datafusion需要单独部署在一个Instance上24小时运行，这个Instance企业版的收费大概一小时要几美元。另外Pipeline运行的时候会调度Dataproc的Instance，这里会产生额外的费用。

方案二的优点是可以灵活的通过Spark代码来完成各种需求。缺点也是比较贵，因为Dataproc是基于Hadoop集群的，需要有Zookeeper, driver和executor这几个VM。如果采用K8S集群，则Spark operator也是需要单独24小时运行在一个pod上，另外还有额外的driver, executor的Pod需要调度执行。

方案三是综合考虑最优的方案，因为Beam的代码是提供了一个通用的流批处理框架，可以运行在Spark,Flink,Dataflow等引擎上，而Dataflow是Google提供的一个优秀的引擎，在运行任务时，Dataflow按需调度VM来运行，只收取运行时的费用。

因此，对于我的这个业务场景，使用方案三是最合适的。下面我将介绍一下整个实现的过程。

Beam批处理任务的实现

在Dataflow的官方Template里面，有一个消费Kafka数据写入到Bigquery的例子，但是这个是流处理方式实现的，对于我的业务场景来说，并不需要这么实时的处理数据，只需要定期消费即可，因此用批处理的方式更合适，这样也能大幅节约费用。

Beam的Kafka I/O connector是默认处理的数据是无边界的，即流式数据。要以批处理的方式来处理，需要调用withStartReadTime和withStopReadTime两个方法获取要读取的Kafka topic的start和end offset，这样就可以把数据转换为有边界数据。调用这两个方法需要注意的是，如果Kafka没有任何一条消息的时间戳是大于等于这个时间戳的话，那么会报错，因此我们需要确定一下具体的时间戳。

以下的代码是检查Kafka消息的所有分区是否存在消息的时间戳是大于我们指定的时间戳，如果不存在的话，那么我们需要找出这些分区里面的最晚时间戳里面的最早的一个。例如Topic有3个分区，要指定的时间戳是1697289783000，但是3个分区里面的所有消息都小于这个时间戳，因此我们需要分别找出每个分区里面的消息的最晚的时间戳，然后取这3个分区的最晚时间戳里面最早的那个，作为我们的指定时间戳。

public class CheckKafkaMsgTimestamp {private static final Logger LOG = LoggerFactory.getLogger(CheckKafkaMsgTimestamp.class);public static KafkaResult getTimestamp(String bootstrapServer, String topic, long startTimestamp, long stopTimestamp) {long max_timestamp = stopTimestamp;long max_records = 5L;Properties props = new Properties();props.setProperty("bootstrap.servers", bootstrapServer);props.setProperty("group.id", "test");props.setProperty("enable.auto.commit", "true");props.setProperty("auto.commit.interval.ms", "1000");props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// Get all the partitions of the topicint partition_num = consumer.partitionsFor(topic).size();HashMap<TopicPartition, Long> search_map = new HashMap<>();ArrayList<TopicPartition> tp = new ArrayList<>();for (int i=0;i<partition_num;i++) {search_map.put(new TopicPartition(topic, i), stopTimestamp);tp.add(new TopicPartition(topic, i));}// Check if message exist with timestamp greater than search timestampBoolean flag = true;ArrayList<TopicPartition> selected_tp = new ArrayList<>();//LOG.info("Start to check the timestamp {}", stopTimestamp);Map<TopicPartition, OffsetAndTimestamp> results = consumer.offsetsForTimes(search_map);for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : results.entrySet()) {OffsetAndTimestamp value = entry.getValue();if (value==null) {   //there is at least one partition don't have timestamp greater or equal to the stopTimeflag = false;break;}}// Get the latest timestamp of all partitions if the above check result is false// Note the timestamp is the earliest of all the partitions. if (!flag) {max_timestamp = 0L;consumer.assign(tp);Map<TopicPartition, Long> endoffsets = consumer.endOffsets(tp);for (Map.Entry<TopicPartition, Long> entry : endoffsets.entrySet()) {Long temp_timestamp = 0L;int record_count = 0;TopicPartition t = entry.getKey();long offset = entry.getValue();if (offset < 1) {LOG.warn("Can not get max_timestamp as partition has no record!");continue;}consumer.assign(Arrays.asList(t));consumer.seek(t, offset>max_records?offset-5:0);Iterator<ConsumerRecord<String, String>> records = consumer.poll(Duration.ofSeconds(2)).iterator();while (records.hasNext()) {record_count++;ConsumerRecord<String, String> record = records.next();LOG.info("Topic: {}, Record Timestamp: {}, recordcount: {}", t, record.timestamp(), record_count);if (temp_timestamp == 0L || record.timestamp() > temp_timestamp) {temp_timestamp = record.timestamp();}}//LOG.info("Record count: {}", record_count);if (temp_timestamp > 0L && temp_timestamp > startTimestamp) {if (max_timestamp == 0L || max_timestamp > temp_timestamp) {max_timestamp = temp_timestamp;}selected_tp.add(t);LOG.info("Temp_timestamp {}", temp_timestamp);LOG.info("Selected topic partition {}", t);LOG.info("Partition offset {}", consumer.position(t));//consumer.seek(t, -1L);}}} else {selected_tp = tp;}consumer.close();LOG.info("Max Timestamp: {}", max_timestamp);return new KafkaResult(max_timestamp, selected_tp);}
}

调用以上代码，我们可以获取要选择的分区以及对应的时间戳。利用这两个信息，我们就可以把指定时间范围内的Kafka数据转换为有边界数据了。以下是Beam建立Pipeline并处理数据，然后写入到Bigquery的代码：

KafkaResult checkResult = CheckKafkaMsgTimestamp.getTimestamp(options.getBootstrapServer(), options.getInputTopic(), start_read_time, stop_read_time);
stop_read_time = checkResult.max_timestamp;
ArrayList<TopicPartition> selected_tp = checkResult.selected_tp;PCollection<String> input = pipeline.apply("Read messages from Kafka",KafkaIO.<String, String>read().withBootstrapServers(options.getBootstrapServer()).withKeyDeserializer(StringDeserializer.class).withValueDeserializer(StringDeserializer.class).withConsumerConfigUpdates(ImmutableMap.of("group.id", "telematics_statistic.app", "enable.auto.commit", true)).withStartReadTime(Instant.ofEpochMilli(start_read_time)).withStopReadTime(Instant.ofEpochMilli(stop_read_time)).withTopicPartitions(selected_tp).withoutMetadata()).apply("Get message contents", Values.<String>create());PCollectionTuple msgTuple = input.apply("Filter message", ParDo.of(new DoFn<String, TelematicsStatisticsMsg>() {@ProcessElementpublic void processElement(@Element String element, MultiOutputReceiver out) {TelematicsStatisticsMsg msg = GSON.fromJson(element, TelematicsStatisticsMsg.class);if (msg.timestamp==0 || msg.vin==null) {out.get(otherMsgTag).output(element);} else {if (msg.timestamp<start_process_time || msg.timestamp>=stop_process_time) {out.get(otherMsgTag).output(element);} else {out.get(statisticsMsgTag).output(msg);}}}}).withOutputTags(statisticsMsgTag, TupleTagList.of(otherMsgTag))); // Get the filter out msg
PCollection<TelematicsStatisticsMsg> statisticsMsg = msgTuple.get(statisticsMsgTag);
// Save the raw records to Bigquery
statisticsMsg.apply("Convert raw records to BigQuery TableRow", MapElements.into(TypeDescriptor.of(TableRow.class)).via(TelematicsStatisticsMsg -> new TableRow().set("timestamp", Instant.ofEpochMilli(TelematicsStatisticsMsg.timestamp).toString()).set("vin", TelematicsStatisticsMsg.vin).set("service", TelematicsStatisticsMsg.service).set("type", TelematicsStatisticsMsg.messageType))).apply("Save raw records to BigQuery", BigQueryIO.writeTableRows().to(options.getStatisticsOutputTable()).withSchema(new TableSchema().setFields(Arrays.asList(new TableFieldSchema().setName("timestamp").setType("TIMESTAMP"),new TableFieldSchema().setName("vin").setType("STRING"),new TableFieldSchema().setName("service").setType("STRING"),new TableFieldSchema().setName("type").setType("STRING")))).withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED).withWriteDisposition(WriteDisposition.WRITE_APPEND));PipelineResult result = pipeline.run();
try {result.getState();result.waitUntilFinish();
} catch (UnsupportedOperationException e) {// do nothing
} catch (Exception e) {e.printStackTrace();
}

需要注意的是，每次处理任务完成后，我们需要把当前的stopReadTime记录下来，下次任务运行的时候把这个时间戳作为startReadTime，这样可以避免某些情况下的数据缺失读取的问题。这个时间戳我们可以把其记录在GCS的bucket里面。这里略过这部分代码。

提交Dataflow任务

之后我们就可以调用Google的Cloud Build功能来把代码打包为Flex Template

首先在Java项目中运行mvn clean package，打包jar文件

然后在命令行中设置以下环境变量：

export TEMPLATE_PATH="gs://[your project ID]/dataflow/templates/telematics-pipeline.json" 
export TEMPLATE_IMAGE="gcr.io/[your project ID]/telematics-pipeline:latest" 
export REGION="us-west1"

之后运行gcloud build的命令来构建镜像：

gcloud dataflow flex-template build $TEMPLATE_PATH --image-gcr-path "$TEMPLATE_IMAGE" --sdk-language "JAVA" --flex-template-base-image "gcr.io/dataflow-templates-base/java17-template-launcher-base:20230308_RC00" --jar "target/telematics-pipeline-1.0-SNAPSHOT.jar" --env FLEX_TEMPLATE_JAVA_MAIN_CLASS="com.example.TelematicsBatch"

最后就可以调用命令来提交任务执行了：

gcloud dataflow flex-template run "analytics-pipeline-`date +%Y%m%d-%H%M%S`" --template-file-gcs-location "$TEMPLATE_PATH" --region "us-west1" --parameters ^~^bootstrapServer="kafka-1:9094,kafka-2:9094"~statisticsOutputTable="youprojectid:dataset.tablename"~serviceAccount="xxx@projectid.iam.gserviceaccount.com"~region="us-west1"~usePublicIps=false~runner=DataflowRunner~subnetwork="XXXX"~tempLocation=gs://bucketname/temp/~startTime=1693530000000~stopTime=1697216400000~processStartTime=1693530000000~processStopTime=1697216400000

如果我们需要任务自动定期执行，还可以在dataflow里面import一个Pipeline，用之前指定的Template_path来导入。然后设置任务的定期周期和启动时间即可，非常方便。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/104920.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

MongoDB的作用和安装方法

MongoDB的作用和安装方法

MongoDB是一种非关系型数据库&#xff0c;其作用是存储和管理非结构化数据&#xff0c;例如文档、图像和视频等多媒体数据。它有以下几个特点&#xff1a; 数据存储的格式是类似JSON的文档格式&#xff0c;易于理解、存储和查询。可扩展性强&#xff0c;可以在多个服务器上分布…
阅读更多...
[资源推荐] 复旦大学张奇老师科研分享

[资源推荐] 复旦大学张奇老师科研分享

刷B站的时候首页给我推了这个&#xff1a;【直播回放】复旦大学张奇教授亲授&#xff1a;人工智能领域顶会论文的发表指南先前也散漫地读了些许论文&#xff0c;但没有在一些宏观的方法论下去训练&#xff0c;读的时候能感觉出一些科研的套路&#xff0c;论文写作的套路&#x…
阅读更多...
C++算法：图中的最短环

C++算法：图中的最短环

题目 现有一个含 n 个顶点的 双向 图&#xff0c;每个顶点按从 0 到 n - 1 标记。图中的边由二维整数数组 edges 表示&#xff0c;其中 edges[i] [ui, vi] 表示顶点 ui 和 vi 之间存在一条边。每对顶点最多通过一条边连接&#xff0c;并且不存在与自身相连的顶点。 返回图中 …
阅读更多...
Kafka保证消息幂等以及解决方案

Kafka保证消息幂等以及解决方案

1、幂等的基本概念 幂等简单点讲&#xff0c;就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的&#xff0c;不会产生任何副作用。幂等分很多种&#xff0c;比如接口的幂等、消息的幂等&#xff0c;它是分布式系统设计时必须要考虑的一个方面。 查询操作(天然幂等…
阅读更多...
【VR开发】【Unity】0-课程简介和概述

【VR开发】【Unity】0-课程简介和概述

【说明】 这是我录制的一套VR基础开发课程的文字版本&#xff0c;更加便于快速参考。 应大家在后台所提的需求&#xff0c;从今天开始&#xff0c;我计划带给大家一套完整达40课时的VR开发基础课程。 在开始学习前需要注意如下几点&#xff1a; 本教程基于Unity2022.2.1f1版…
阅读更多...
SqlServer安装教程

SqlServer安装教程

百度网盘地址: 链接&#xff1a;https://pan.baidu.com/s/1ntqoK9uVc6fBVTm7twh8kw 提取码&#xff1a;grdt 安装: 双击:SQLEXPRADV_x64_CHS.exe ,等待;点击计划,系统配置检查器,根据要求修改(我被要求重启了)点击安装,全新SQL Server独立安装或向现有安装添加功能,接受功能选…
阅读更多...
2023年09月 C/C++（七级）真题解析#中国电子学会#全国青少年软件编程等级考试

2023年09月 C/C++（七级）真题解析#中国电子学会#全国青少年软件编程等级考试

C/C编程&#xff08;1~8级&#xff09;全部真题・点这里 Python编程&#xff08;1~6级&#xff09;全部真题・点这里 第1题&#xff1a;红与黑 有一间长方形的房子&#xff0c;地上铺了红色、黑色两种颜色的正方形瓷砖。你站在其中一块黑色的瓷砖上&#xff0c;只能向相邻的黑色…
阅读更多...
线性筛的简单证明

线性筛的简单证明

原理 线性筛是一种可以在线性时间内将素数筛选出来的算法&#xff0c;其中的主要思想在于保证合数只会被它的最小质因数筛掉并且筛掉一次。 代码 下面是线性筛的算法CPP实现&#xff1a; vector<int> generate_primes_linear_time(int n) {vector<int> lp(n 1…
阅读更多...
kafka详解(三)

kafka详解(三)

2.2 Kafka命令行操作 2.2.1 主题命令行操作 1&#xff09;查看操作主题命令参数 [aahadoop102 kafka]$ bin/kafka-topics.sh2&#xff09;查看当前服务器中的所有topic (配置了环境变量不需要写bin/) [aahadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop10…
阅读更多...
SpringBoot-黑马程序员-学习笔记（五）

SpringBoot-黑马程序员-学习笔记（五）

74.自定义bean属性绑定以及第三方bean属性绑定 自定义bean属性绑定 1.自定义一个bean Data Component public class ServerConfig {private String ipAddress;private int port;private long timeout; } 2.在yml配置文件中中定义一组值 3.在bean中进行属性绑定 加上这个注…
阅读更多...
mysql中的几种排名函数

mysql中的几种排名函数

mysql中的排名函数 mysql里面的排名函数&#xff0c;涉及有以下几个&#xff1a; rank()、dense_rank()、row_number() 1、rank() 函数 RANK() OVER (PARTITION BY <expression>[{,<expression>...}]ORDER BY <expression> [ASC|DESC], [{,<expression…
阅读更多...
【Nginx32】Nginx学习：随机索引、真实IP处理与来源处理模块

【Nginx32】Nginx学习：随机索引、真实IP处理与来源处理模块

Nginx学习&#xff1a;随机索引、真实IP处理与来源处理模块 完成了代理这个大模块的学习&#xff0c;我们继续其它 Nginx 中 HTTP 相关的模块学习。今天的内容都比较简单&#xff0c;不过最后的来源处理非常有用&#xff0c;可以帮我们解决外链问题。另外两个其实大家了解一下就…
阅读更多...
C#开发的OpenRA游戏之金钱系统（1）

C#开发的OpenRA游戏之金钱系统（1）

C#开发的OpenRA游戏之金钱系统（1） 设计一个游戏，肯定要有一个唯一的资源，用这个资源来管理整个游戏的进度，以及相互争夺的焦点。在OpenRA里，就是使用矿产资源。所以在地图上分布几个矿场，玩家就需要相互争夺矿场，谁开采多谁就更有钱，谁有钱了就可以升级更好的科技，以…
阅读更多...
Linux Kernel 4.13 RC6发布：正式版9月3日发布

Linux Kernel 4.13 RC6发布：正式版9月3日发布

美国当地时间上周末&#xff0c;大神Linus Torvalds发布了Linux Kernel 4.13内核的又一候选版本。上周发布的RC5版本更新幅度也要比上上周的RC4要小&#xff0c;Linus Torvalds表示本周发布的RC6版本属于常规更新&#xff0c;在过去一周的开发过程中并没有出现任何意外。RC6版本…
阅读更多...
Spring MVC 十一：@EnableWebMvc

Spring MVC 十一：@EnableWebMvc

我们从两个角度研究EnableWebMvc&#xff1a; EnableWebMvc的使用EnableWebMvc的底层原理 EnableWebMvc的使用 EnableWebMvc需要和java配置类结合起来才能生效&#xff0c;其实Spring有好多Enablexxxx的注解&#xff0c;其生效方式都一样&#xff0c;通过和Configuration结合…
阅读更多...
Linux 64位 C++协程池原理分析及代码实现

Linux 64位 C++协程池原理分析及代码实现

导语 本文介绍了协程的作用、结构、原理&#xff0c;并使用C和汇编实现了64位系统下的协程池。文章内容避免了协程晦涩难懂的部分&#xff0c;用大量图文来分析原理&#xff0c;适合新手阅读学习。 GitHub源码 1. Web服务器问题 现代分布式Web后台服务逻辑通常由一系列RPC请…
阅读更多...
【java学习—七】单继承和多层继承（30）

【java学习—七】单继承和多层继承（30）

文章目录 1. 相关概念2. 从代码中理解 1. 相关概念 Java 只支持单继承&#xff0c;不允许多重继承&#xff1a; &#xff08;1&#xff09;一个子类只能有一个父类 &#xff08;2&#xff09;一个父类可以派生出多个子类      举例区分&#xff1a; class SubDemo extend…
阅读更多...
Hermes - 指尖上的智慧：自定义问答系统的崭新世界

Hermes - 指尖上的智慧：自定义问答系统的崭新世界

在希腊神话中&#xff0c;有一位智慧与消息的传递者神祇&#xff0c;他就是赫尔墨斯&#xff08;Hermes&#xff09;。赫尔墨斯是奥林匹斯众神中的一员&#xff0c;传说他是乌尔阿努斯&#xff08;Uranus&#xff09;和莫伊拉&#xff08;Maia&#xff09;的儿子&#xff0c;同…
阅读更多...
Git纯操作版 项目添加和提交、SSH keys添加、远程仓库控制、冲突解决、IDEA连接使用

Git纯操作版 项目添加和提交、SSH keys添加、远程仓库控制、冲突解决、IDEA连接使用

Git 文章目录 Git项目简单克隆通用操作添加和提交回滚分支变基分支优选 远程项目推送认证抓取、拉取和冲突解决 IEDA类软件连接 最近学原理学的快头秃了&#xff0c;特此想出点不讲原理的纯操作版&#xff0c;不过还是放个图吧 项目简单克隆 git在本人日常中最重要的功能还是…
阅读更多...
Linux中怎么启动Zookeeper

Linux中怎么启动Zookeeper

首先进入Zookeeper安装目录下的bin目录 比如&#xff1a; cd /root/zookeeper-3.4.9/bin 然后在此目录下执行命令。 1. 启动Zookeeper Server端 ./zkServer.sh start 2.启动Zookeeper Client端 ./zkCli.sh 启动Zookeeper Client端后如下&#xff1a;
阅读更多...
最新文章

Copyright @ 2022~2023