表流读取和写入

限制输入的速度

以下选项可以控制micro-batches:

• maxFilesPerTrigger:有多少新文件被认为是在每个micro-batch。默认值是1000。

• maxBytesPerTrigger在每个micro-batch:多少数据被处理。这个选项设置一个“软马克斯”,这意味着一个批处理过程大约这个过程的数据量,可能超过极限为了使流查询前进情况下的最小输入单位超过这个极限。如果你使用Trigger.Once为你流,此选项将被忽略。这不是默认设置。

如果你使用maxBytesPerTrigger结合maxFilesPerTrigger,micro-batch过程数据,直到maxFilesPerTrigger或maxBytesPerTrigger达到极限。

河三角洲湖变化数据捕获(CDC)饲料

三角洲湖改变数据提要三角洲的记录更改表,包括更新和删除。启用时,你可以从改变流数据饲料和编写逻辑处理插入、更新和删除操作到下游的表中。尽管变化数据提要数据输出三角洲表描述略有不同,这提供了一种解决方案宣传下游表的增量变化大奖章架构。

忽略更新和删除

结构化流不处理不是一个附加的输入,将抛出一个异常如果发生任何修改在桌子上被用作源。有两个主要的策略来处理变化,不能自动下游传播:

• 您可以删除输出从一开始就和检查点和重启流。

• 你可以设置这两个选择:

  • ignoreDeletes:忽略事务删除数据分区的边界。

  • ignoreChanges:如果文件必须被重写处理文档更新源表中的数据修改操作,比如更新,合并成,删除(分区),或覆盖。不变行可能仍在释放,因此你的下游消费者应该能够处理重复。删除不是下游传播。ignoreChanges包容ignoreDeletes。因此如果你使用ignoreChanges,你流将不会被删除或更新源表。

火花。readStream。格式(“δ”)。选项(“ignoreDeletes”,“真正的”)。负载(“/ tmp /δ/ user_events”)

火花。readStream。格式(“δ”)。选项(“ignoreChanges”,“真正的”)。负载(“/ tmp /δ/ user_events”)

指定初始位置

您可以使用以下选项指定的起点三角洲湖流源没有处理整个表。

• startingVersion:从三角洲湖版本。所有表变化从这个版本(包容)将读取流源。你可以获取提交的版本版本列的描述历史命令的输出。

  在砖运行时的7.4及以上,只返回最新的变化,指定最新的。

• startingTimestamp:从时间戳。所有表更改后承诺或时间戳(包容)将读取流源。之一:

  • 时间戳字符串。例如,“2019 - 01 - 01 t00:00:00.000z”。

  • 一个日期字符串。例如,“2019-01-01”。

你不能同时设置两个选项;你可以只使用其中的一个。他们只有当开始一个新的流媒体查询生效。如果一个流媒体查询已经开始和检查点的进展记录,这些选项将被忽略。

火花。readStream。格式(“δ”)。选项(“startingVersion”,“5”)。负载(“/ tmp /δ/ user_events”)

火花。readStream。格式(“δ”)。选项(“startingTimestamp”,“2018-10-18”)。负载(“/ tmp /δ/ user_events”)

没有数据被删除过程初始快照

使用增量表作为流源时,数据的查询过程首先出现在桌子上。三角洲表在这个版本被称为初始快照。默认情况下,三角洲表的数据文件处理基于文件最后修改。然而,最后修改时间并不一定代表记录事件的时间顺序。

在有状态流查询定义水印,处理文件的修改时间会导致记录处理错误的订单。这可能导致晚期事件的记录删除水印。

你可以避免数据下降问题通过支持下列选项:

• withEventTimeOrder:初始快照是否应该处理事件的时间顺序。

启用事件时间顺序后,初始快照数据的事件时间范围分为桶。每个微一桶的批处理流程过滤数据的时间范围内。maxFilesPerTrigger和maxBytesPerTrigger配置选项仍适用于控制microbatch大小,但只在一个近似的方法由于处理的本质。

下图显示了这个过程:

明显的这一特性的信息:

• 数据下降问题只有当初始增量快照的状态在默认顺序流查询处理。

• 你不能改变withEventTimeOrder一旦流查询开始时初始快照还是正在处理。重新启动,withEventTimeOrder改变,你需要删除检查点。

• 如果您正在运行一个启用了withEventTimeOrder流查询,你不能降级到DBR版本不支持这个功能,直到初始快照处理完成。如果你需要降级,你可以等待初始快照完成,查询或删除检查点和重启。

• 不支持这个功能在以下常见场景:

  • 事件时间列是一个生成的列有non-projectionδ源和水印之间的转换。

  • 有一个水印,不止一个δ源在流查询。

• 启用事件时间顺序后,三角洲初始快照处理的性能可能较慢。

• 每一批微扫描初始快照来过滤数据对应的事件时间范围内。加快过滤操作,建议使用一个δ源列事件时间,以便数据不可以应用(检查数据不与z顺序索引三角洲湖在适当的时候)。另外,表分区沿着事件时间列可以进一步加快处理。您可以检查火花UI,看看有多少δ特定微批处理文件扫描。

火花。readStream。格式(“δ”)。选项(“withEventTimeOrder”,“真正的”)。负载(“/ tmp /δ/ user_events”)。withWatermark(“event_time”,“10秒”)

指标

你可以找到的字节数和文件数量有待处理流媒体查询流程随着numBytesOutstanding和numFilesOutstanding指标。额外的指标包括:

• numNewListedFiles:列出的三角洲湖文件数量来计算这批的积压。

  • backlogEndOffset:表版本用来计算积压。

如果您正在运行流在一个笔记本,你可以看到这些度量标准下原始数据选项卡中流仪表板查询进展:

{“源”:({“描述”:“DeltaSource(文件/道路/ /源):“,“指标”:{“numBytesOutstanding”:“3456”,“numFilesOutstanding”:“8”},}]}

Append模式

默认情况下,流在附加模式下运行,向表添加新记录。

您可以使用路径的方法:

(事件。writeStream。格式(“δ”)。outputMode(“添加”)。选项(“checkpointLocation”,“/ tmp /δ/ _checkpoints /”)。开始(“/δ/事件”))

事件。writeStream。格式(“δ”)。outputMode(“添加”)。选项(“checkpointLocation”,“/ tmp /δ/事件/ _checkpoints /”)。开始(“/ tmp /δ/事件”)

或者是toTable方法在火花3.1和更高版本(砖运行时的8.3及以上),如下所示。(在火花之前版本3.1(砖运行时8.2及以下),使用表方法。)

(事件。writeStream。格式(“δ”)。outputMode(“添加”)。选项(“checkpointLocation”,“/ tmp /δ/事件/ _checkpoints /”)。toTable(“事件”))

事件。writeStream。outputMode(“添加”)。选项(“checkpointLocation”,“/ tmp /δ/事件/ _checkpoints /”)。toTable(“事件”)

完整的模式

您还可以使用结构化流与每一批替换整个表。用例的一个例子是使用聚合计算一个总结:

(火花。readStream。格式(“δ”)。负载(“/ tmp /δ/事件”)。groupBy(“customerId”)。数()。writeStream。格式(“δ”)。outputMode(“完整的”)。选项(“checkpointLocation”,“/ tmp /δ/ eventsByCustomer / _checkpoints /”)。开始(“/ tmp /δ/ eventsByCustomer”))

火花。readStream。格式(“δ”)。负载(“/ tmp /δ/事件”)。groupBy(“customerId”)。数()。writeStream。格式(“δ”)。outputMode(“完整的”)。选项(“checkpointLocation”,“/ tmp /δ/ eventsByCustomer / _checkpoints /”)。开始(“/ tmp /δ/ eventsByCustomer”)

前面的例子不断更新一个表,其中包含客户总数量的事件。

更宽松的延迟需求的应用程序,你可以节省计算资源与一次性触发器。使用这些更新总结聚合表在给定的时间表,只处理新数据到达自上次更新。

例子

app_id=…#一个独一无二的字符串用作应用程序ID。defwriteToDeltaLakeTableIdempotent(batch_df,batch_id):batch_df。写。格式(…)。选项(“txnVersion”,batch_id)。选项(“txnAppId”,app_id)。保存(…)#位置1batch_df。写。格式(…)。选项(“txnVersion”,batch_id)。选项(“txnAppId”,app_id)。保存(…)#位置2

瓦尔appId=…/ /一个独一无二的字符串用作应用程序ID。streamingDF。writeStream。foreachBatch{(batchDF:DataFrame,batchId:长)= >batchDF。写。格式(…)。选项(“txnVersion”,batchId)。选项(“txnAppId”,appId)。保存(…)/ /位置1batchDF。写。格式(…)。选项(“txnVersion”,batchId)。选项(“txnAppId”,appId)。保存(…)/ /位置2}