在Databricks上使用結構化流的示例

使用Python

這個例子展示了如何使用streamingDataFrame.writeStream.foreach ()用Python寫到DynamoDB。第一步獲取DynamoDB boto資源。編寫此示例是為了使用access_key和secret_key，但Databricks建議您使用使用實例配置文件安全訪問S3桶．

1. 定義一些幫助器方法來創建用於運行示例的DynamoDB表。

   table_name＝“PythonForeachTest”defget_dynamodb()：進口boto3access_key＝“< >訪問關鍵”secret_key＝“<密鑰>”地區＝“<地區名稱>”返回boto3．資源（“dynamodb”，aws_access_key_id＝access_key，aws_secret_access_key＝secret_key，region_name＝地區）defcreateTableIfNotExists()：“‘如果DynamoDB表不存在，則創建它。這必須在Spark驅動程序上運行，而不是在foreach內部。“‘dynamodb＝get_dynamodb（）existing_tables＝dynamodb．元．客戶端．list_tables() (的表名］如果table_name不在existing_tables：打印（“創建表% s＂％table_name）表格＝dynamodb．create_table（的表＝table_name，KeySchema＝［｛“AttributeName”：“關鍵”，“KeyType”：“希”｝),AttributeDefinitions＝［｛“AttributeName”：“關鍵”，“AttributeType”：“年代”｝),ProvisionedThroughput＝｛“ReadCapacityUnits”：5，“WriteCapacityUnits”：5｝）打印（“等待餐桌準備好”）表格．元．客戶端．get_waiter（“table_exists”）．等待（的表＝table_name）

2. 定義寫入DynamoDB並調用它們的類和方法foreach．中指定自定義邏輯有兩種方法foreach．

   • 使用函數:這是一種簡單的方法，可用於每次寫入一行。但是，每次調用都會初始化客戶端/連接來寫入一行。

     defsendToDynamoDB_simple（行)：“‘函數將一行發送到DynamoDB。當與' foreach '一起使用時，這個方法將在執行器中被調用使用生成的輸出行。“‘#在執行器中創建客戶端對象#不要使用在驅動中創建的客戶端對象dynamodb＝get_dynamodb（）dynamodb．表格（table_name）．put_item（項＝｛“關鍵”：str（行［“關鍵”]),“數”：行［“數”］})

   • 使用帶有開放，過程,關閉方法:這允許更有效的實現，其中客戶端/連接被初始化並可以寫入多行。

     類SendToDynamoDB_ForeachWriter：“‘類向DynamoDB發送一組行。當與' foreach '一起使用時，這個類的副本將被用於寫入執行器中的多行。參見' DataStreamWriter.foreach '的python文檔為更多的細節。“‘def開放（自我，partition_id，epoch_id)：#在準備發送多行時首先調用。#將所有初始化代碼放在open()中，這樣就可以創建一個fresh在執行器中初始化這個類的副本#將被調用。自我．dynamodb＝get_dynamodb（）返回真正的def過程（自我，行)：在open()被調用之後，每一行都會調用這個函數。這個實現一次發送一行。#要進一步增強，請聯係Spark+DynamoDB連接器#團隊:https://github.com/audienceproject/spark-dynamodb自我．dynamodb．表格（table_name）．put_item（項＝｛“關鍵”：str（行［“關鍵”]),“數”：行［“數”］})def關閉（自我，犯錯)：在處理完所有行之後調用。如果犯錯：提高犯錯

3. 調用foreach在使用上述函數或對象的流查詢中。

   從pyspark.sql.functions進口＊火花．相依．集（“spark.sql.shuffle.partitions”，“1”）查詢＝（火花．readStream．格式（“速度”）．負載（）．selectExpr（"值% 10為鍵"）．groupBy（“關鍵”）．數（）．toDF（“關鍵”，“數”）．writeStream．foreach（SendToDynamoDB_ForeachWriter())#.foreach(sendToDynamoDB_simple) //選擇一個或另一個．outputMode（“更新”）．開始（））

使用Scala

這個例子展示了如何使用streamingDataFrame.writeStream.foreach ()在Scala中寫入DynamoDB。

要運行這個程序，您必須創建一個DynamoDB表，它隻有一個名為“value”的字符串鍵。

1. 類的實現ForeachWriter執行寫操作的接口。

   進口org．apache．火花．sql．｛ForeachWriter，行｝進口com．amazonaws．AmazonServiceException進口com．amazonaws．身份驗證．＿進口com．amazonaws．服務．dynamodbv2．AmazonDynamoDB進口com．amazonaws．服務．dynamodbv2．AmazonDynamoDBClientBuilder進口com．amazonaws．服務．dynamodbv2．模型．AttributeValue進口com．amazonaws．服務．dynamodbv2．模型．ResourceNotFoundException進口java．跑龍套．ArrayList進口scala．集合．JavaConverters．＿類DynamoDbWriter擴展ForeachWriter［行］｛私人瓦爾的表＝“<表名>”私人瓦爾accessKey＝aws訪問密鑰“< >”私人瓦爾secretKey＝“< > aws密鑰”私人瓦爾regionName＝地區“< >”//隻有在調用open()時才會惰性地初始化懶惰的瓦爾ddb＝AmazonDynamoDBClientBuilder．標準（）．withCredentials（新AWSStaticCredentialsProvider（新BasicAWSCredentials（accessKey，secretKey))）．withRegion（regionName）．構建（）////在準備發送多行時首先調用。//把所有的初始化代碼放在open()中，這樣一個fresh//在執行器中初始化這個類的副本//將被調用。//def開放（partitionId：長，epochId：長）＝｛ddb//強製客戶端初始化真正的｝////在open()被調用之後，每一行都調用這個函數。//此實現每次發送一行。//一個更有效的實現是一次發送多個行。//def過程（行：行）＝｛瓦爾rowAsMap＝行．getValuesMap（行．模式．字段名）瓦爾dynamoItem＝rowAsMap．mapValues｛v：任何= >新AttributeValue（v．toString）}。asJavaddb．putItem（的表，dynamoItem）｝////在處理完所有行之後調用。//def關閉（errorOrNull：Throwable）＝｛ddb．關閉（）｝｝

2. 使用DynamoDbWriter將速率流寫入DynamoDB。

   火花．readStream．格式（“速度”）．負載（）．選擇（“價值”）．writeStream．foreach（新DynamoDbWriter）．開始（）