使用foreachBatch编写任意数据汇

重用现有的批处理的数据源

对于许多存储系统来说,可能没有一个流水槽可用,但可能已经存在一个数据作家对于批处理查询。使用foreachBatch (),您可以使用批处理数据作家每个micro-batch的输出。下面是一些例子:

• 卡桑德拉Scala的例子

• Azure突触Python示例分析

许多其他的批处理数据源可以使用从foreachBatch ()。

写入多个位置

如果你想查询的输出流写入多个位置,那么您可以简单地编写多次输出DataFrame /数据集。然而,每个试图写会导致输出数据重新计算(包括可能的重读输入数据)。为了避免重新计算,你应该缓存输出DataFrame /数据集,写多个位置,然后uncache它。这是一个大纲。

streamingDF。writeStream。foreachBatch{(batchDF:DataFrame,batchId:长)= >batchDF。坚持()batchDF。写。格式(…)。保存(…)/ /位置1batchDF。写。格式(…)。保存(…)/ /位置2batchDF。unpersist()}

申请额外DataFrame操作

许多DataFrame和数据集操作不支持流媒体DataFrames因为火花不支持生成增量计划在这些情况下。使用foreachBatch ()你可以把这些操作在每个micro-batch输出。例如,您可以使用foreachBath ()和SQL合并成写操作的输出流聚合成三角洲表更新模式。看到更多的细节合并成。

一个空dataframe可以调用foreachBatch ()和用户代码需要有弹性,以便正确操作。一个例子所示:

。foreachBatch((outputDf:DataFrame,报价:长)= >{/ /流程有效的数据帧如果(!outputDf。isEmpty){/ /业务逻辑}})。开始()

使用Scala或Java

在Scala或Java,您扩展类ForeachWriter:

datasetOfString。writeStream。foreach(新ForeachWriter(字符串]{def开放(partitionId:长,版本:长):布尔={/ /打开连接}def过程(记录:字符串)={/ /写字符串连接}def关闭(errorOrNull:Throwable):单位={/ /关闭连接}})。开始()

使用Python

在Python中,您可以调用foreach在两个方面:在一个函数或一个对象。函数提供了一种简单的方式来表达你的处理逻辑,但不允许您删除处理失败时生成的数据导致一些输入数据的后处理。这种情况下你必须指定一个对象的处理逻辑。

• 的函数行作为输入。

  defprocessRow(行):/ /写行来存储查询=streamingDF。writeStream。foreach(processRow)。开始()

• 的对象有一个过程方法和可选的开放和关闭方法:

  类ForeachWriter:def开放(自我,partition_id,epoch_id):/ /开放连接。这方法是可选在Python。def过程(自我,行):/ /写行来连接。这方法是不可选在Python。def关闭(自我,错误):/ /关闭的连接。这方法是可选在Python。查询=streamingDF。writeStream。foreach(ForeachWriter())。开始()

执行语义

流媒体查询开始时,火花调用函数或对象的方法在以下方式:

• 这个对象的一个副本负责所有数据查询中生成的一个任务。换句话说,一个实例负责处理一个分区的数据在一个分布式的方式生成。

• 这个对象必须是可序列化的,因为每个任务将得到一个新的serialized-deserialized提供对象的副本。因此,强烈建议任何写作初始化数据(例如,打开一个连接或启动一个事务)完成后调用open ()方法,它意味着任务准备生成数据。

• 的生命周期方法如下:

  为每个分区partition_id:

  对于每一批/时代的流数据epoch_id:

  方法打开(partitionIdepochId)被称为。

  如果打开(…)返回true,分区和批处理中的每一行/时代,方法过程(行)被称为。

  方法关上(错误)被称为时遇到以下错误(如果有的话)处理的行。

• 的close ()方法(如果它存在的话)如果一个open ()方法存在并返回成功(不管返回值),除非中间JVM或Python程序崩溃。