亚马逊运动

在时间点上开始阅读

开始阅读的时候,你可以使用一个at_timestamp值initialPosition选择。您指定的值作为一个JSON字符串,如{“at_timestamp”:“06/25/202010:23:45PDT "}。流查询将读取所有更改或在给定的时间戳(包容)。它使用Java的默认格式解析时间戳。您可以显式地指定格式通过提供JSON字符串中的一个额外字段,例如:

(火花。readStream。格式(“运动”)。选项(“streamName”,kinesisStreamName)。选项(“地区”,kinesisRegion)。选项(“initialPosition”,”{at_timestamp”:“06/25/2020 10:23:45 PDT”、“格式”:“MM / dd / yyyy HH: MM: ss ZZZ”}”)。选项(“awsAccessKey”,awsAccessKeyId)。选项(“awsSecretKey”,awsSecretKey)。负载())

此外,还有配置从运动控制阅读的吞吐量和延迟。运动源在一个后台线程运行引发工作定期预抓取动作的内存和缓存的数据引发执行人。缓存数据流查询处理每个预取步骤完成后,使数据进行处理。因此,这个预取步骤决定了许多观察到的端到端延时和吞吐量。你可以使用以下选项控制性能。

这里有一些建议配置特定的用例。

ETL从运动到S3

当你执行ETL长期存储,你宁愿有少量的大文件。在这种情况下,您可能希望设置一个大型流触发间隔,例如,5 - 10分钟。此外,您可能想要增加你的maxFetchDuration这样你缓冲大块将要写入处理期间,和增加fetchBufferSize所以你不要停止抓取过早在触发器之间,并开始落后在你流。

低延迟监控和报警

当你有一个报警用例,你会想要更低的延迟。实现:

• 确保只有一个消费者(也就是说,只有你流查询和没有人)的运动流,这样我们就可以优化你的只有流查询获取尽可能快没有跑到运动速度限制。

• 设置选项maxFetchDuration小值(说,200毫秒)开始尽可能快地处理获取数据。

• 设置选项minFetchPeriod来210毫秒获取尽可能经常。

• 设置选项shardsPerTask或配置集群等#核在集群> =2*(#运动碎片)/shardsPerTask。这将确保背景预取和流媒体查询任务可以并行执行任务。

如果你看到,你查询接收数据每5秒,那么很可能你击球动作速度限制。回顾你的配置。

指标

运动报道的毫秒数消费者已经落后于流的开始为每个工作区。你可以得到平均、最小和最大的毫秒数在所有的工作区流查询过程中(https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html reading-metrics-interactively)作为avgMsBehindLatest,maxMsBehindLatest,minMsBehindLatest指标。如果您正在运行流在一个笔记本,你可以看到这些度量标准下原始数据流查询进展仪表板选项卡:

{“源”:({“描述”:“KinesisV2(流)”,“指标”:{“avgMsBehindLatest”:“32000.0”,“maxMsBehindLatest”:“32000”,“minMsBehindLatest”:“32000”},}]}