亞馬遜運動

在時間點上開始閱讀

開始閱讀的時候,你可以使用一個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”},}]}