从 Kinesis 移动数据 -> RDS 使用 Spark 和 AWS Glue 在本地实施答案

【问题标题】：Moving data from Kinesis -> RDS using Spark with AWS Glue implementation locally从 Kinesis 移动数据 -> RDS 使用 Spark 和 AWS Glue 在本地实施
【发布时间】：2020-06-04 14:35:17
【问题描述】：

我有一个在本地运行 AWS Glue 实施的 Spark 项目。

我收听 Kinesis 流，因此当数据以 JSON 格式到达时，我可以正确存储到 S3。我想存储在 AWS RDS 中，而不是存储在 S3 中。

我试过用：

dataFrame.write
          .format("jdbc")
          .option("url","jdbc:mysql://aurora.cluster.region.rds.amazonaws.com:3306/database")
          .option("user","user")
          .option("password","password")
          .option("dbtable","test-table")
          .option("driver","com.mysql.jdbc.Driver")
          .save()

Spark 项目使用 AWS 粘合作业从 Kinesis 流中获取数据。

我想将数据添加到 Aurora 数据库。

失败并出现错误

Caused by: java.sql.SQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL
 server version for the right syntax to use near '-glue-table (`label2` TEXT , `customerid` TEXT , `sales` TEXT , `name` TEXT )' a
t line 1

这是我正在使用的测试数据帧，dataFrame.show()：

+------+----------+-----+--------------------+
|label2|customerid|sales|                name|
+------+----------+-----+--------------------+
| test6|      test| test|streamingtesttest...|
+------+----------+-----+--------------------+

【问题讨论】：

我认为您缺少上述命令中的 MODE 参数。模式可以是追加、覆盖、错误、忽略。请参考此示例：stackoverflow.com/questions/46552161/…
另外我希望您已经将 JSON 转换为具有适当列标题等的数据框。
我添加了 MODE 参数，但仍然失败。我认为数据框还可以。当我打印它（dataFrame.show）时，它打印得很好（在问题中添加）。 @Yuva

标签： apache-spark aws-glue amazon-aurora

【解决方案1】：

使用 Spark DynamicFrame 代替 DataFrame 并使用glueContext sink 发布到Aurora：

所以最终的代码可能是：

lazy val mysqlJsonOption = jsonOptions(MYSQL_AURORA_URI)

//Write to Aurora
val dynamicFrame = DynamicFrame(joined, glueContext)
glueContext.getSink("mysql", mysqlJsonOption).writeDynamicFrame(dynamicFrame)

【讨论】：