Google Cloud dataflow streaming job简介

本文主要是介绍Google Cloud dataflow streaming job简介，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

简单介绍

首先 gcp 的dataflow 是1个ETL 组件, 它是基于Apache beam的

Apache beam 是1个较新的开源ETL 框架。
对于我们常用的ETL tool Spring batch 有下面的区别

spring batch 更偏向batch （后台处理）的ETL，而apache beam 同时支持batch 和 streaming 的ETL，对streaming 的ETL 有更好的支持
spring batch 基于java，apache beam SDK 支持 java, python 和 GO
spring batch 更加轻量级，但是依赖于1个关系数据存储ETL job（配置，历史记录）数据. 而且不需要开发人员去设置. 而 Apache Beam本身并没有内置的作业执行历史记录功能，这些数据需要自己去记录处理（在runner上）
spring batch 的运行环境需要自己搭建，而apache beam 这是1个SDK，它只定义 ETL pipeline的流程，它需要额外的runner去执行

apache beam 暂时支持下面的runner
GCP dataflow
Apache Flink
Apache Spark
AWS data analysic
Java direct runner(调试用）

所以对于日志记录更多地去交给runner去实现。

所以讲， dataflow 只是1个runner ，核心还是apache beam SDK

什么是streaming

在计算领域中，“streaming”（流式处理）是一种数据处理模式，它允许实时处理连续流入的数据，而不是一次性处理静态数据集。
传统的批处理模式是将数据分成固定大小的块（batch），然后对每个批次进行处理。这种方式适用于静态数据集，但对于实时数据流，它可能无法满足实时性和低延迟的要求。
流式处理模式通过连续接收和处理数据流，实现了实时性和低延迟。数据流可以是连续的数据记录、事件流、传感器数据等。流式处理系统会持续地接收数据流，并立即对其进行处理和分析，以产生实时的结果。
流式处理通常具有以下特点：
连续性：数据流是连续不断的，没有明确的开始和结束。处理系统需要实时接收和处理数据流，而不是等待所有数据到达后再进行处理。
实时性：流式处理系统需要尽可能快地处理数据，并产生实时的结果。这对于需要实时决策、监控和反馈的应用程序非常重要。
有限状态：流式处理系统通常使用有限的内存和状态来处理数据流。它们需要在有限的资源下有效地处理无限的数据流。
流式处理可以应用于各种场景，如实时分析、实时监控、实时推荐、欺诈检测等。流式处理框架（如Apache Flink、Apache Kafka Streams、Apache Beam等）提供了方便的工具和API来开发和部署流式处理应用程序。

1个例子

 public void process() {log.info("processing3...");DataflowPipelineOptions options = PipelineOptionsFactory.fromArgs("").as(DataflowPipelineOptions.class);options.setJobName("dataflow-exam3");options.setProject(this.projectId);options.setRegion("europe-west1");options.setTempLocation("gs://jason-hsbc-dataflow/tmp");options.setSubnetwork("regions/europe-west1/subnetworks/subnet-1");options.setNumWorkers(1);options.setStreaming(true);options.setDefaultWorkerLogLevel(DataflowWorkerLoggingOptions.Level.INFO);options.setNumberOfWorkerHarnessThreads(2);//options.setGcpCredential(new File(...));options.setRunner(DataflowRunner.class);log.info(getCurrentAccountName());Pipeline pipeline = Pipeline.create(options);/*** * The effect of using the @UnknownKeyFor annotation is to tell the Apache Beam framework that the PCollection does not need to be grouped or associated based on a specific key.* As a result, Apache Beam can perform more efficient parallel computations and optimizations for operations that do not require key associations.** * When the @NonNull annotation is applied to a PCollection element type, it indicates that elements in the PCollection are not allowed to be null.* This means that when processing data streams, the Apache Beam framework checks elements for non-nullability and issues warnings or errors at compile time to avoid potential null pointer exceptions.the @Initialized annotation informs the compiler that the variable has been properly initialized by marking it at variable declaration time to help detect possible null pointer exceptionsand improve the reliability and readability of the code. However, it should be noted that the @Initialized annotation is only an auxiliary tool, and correct logic and design still need to be ensured in the actual programming process.*/PCollection<@UnknownKeyFor @NonNull @Initialized PubsubMessage> message = pipeline.apply("Read Pub/Sub Messages", PubsubIO.readMessagesWithAttributesAndMessageId().fromSubscription("projects/jason-hsbc/subscriptions/SubscriptionA1"));PCollection<KV<String, String>> combinedMsg = message.apply("Extract", ParDo.of(new ExtractMessageAttributeFn())).apply("appying windowing", Window.into(FixedWindows.of(Duration.standardMinutes(1)))).apply("Group by fileName", GroupByKey.create()).apply("Combine Message",ParDo.of(new CombinedMessagesFn()));combinedMsg.apply("Write to GCS", ParDo.of(new WriteToGCSFn(this.bucketName,this.projectId)));//yupipeline.run().waitUntilFinish();pipeline.run();log.info("processing3... end!");}