数据库上的结构化流模式

使用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 '一起使用时，这个类的副本将用于写入执行程序中的多行。参见python文档中的“DataStreamWriter.foreach”欲知详情。“‘def开放（自我，partition_id，epoch_id)：#在准备发送多行时首先调用。#把所有的初始化代码放在open()里面，这样就可以这个类的拷贝在执行器中初始化#将被调用。自我．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＝""私人瓦尔secretKey＝""私人瓦尔regionName＝地区“< >”//只有在调用open()时才会被惰性地初始化懒惰的瓦尔ddb＝AmazonDynamoDBClientBuilder．标准（）．withCredentials（新AWSStaticCredentialsProvider（新BasicAWSCredentials（accessKey，secretKey)))．withRegion（regionName）．构建（）////在准备发送多行时首先调用。//把所有的初始化代码放在open()里面，这样就可以//在执行程序中初始化这个类的副本，其中open()//将被调用。//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）．开始（）