从选项（来自 rds - mysql）创建动态框架，提供带有 where 子句的自定义查询

Posted 2023-04-15

【中文标题】从选项（来自 rds - mysql）创建动态框架，提供带有 where 子句的自定义查询

【英文标题】：Create dynamic frame from options (from rds - mysql) providing a custom query with where clause

【发布时间】：2020-02-16 18:48:07

【问题描述】：

我想在我的 Glue 作业中从 Aurora-rds mysql 表创建一个 DynamicFrame。我可以使用自定义查询从我的 rds 表创建 DynamicFrame - 有 where 子句吗？ 我不想每次都在我的 DynamicFrame 中读取整个表格，然后再过滤。 看了这个网站，但在这里或其他地方没有找到任何选项，https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-connect.html

构造 JDBC 连接选项

connection_mysql5_options = "url": "jdbc:mysql://:3306/db", “dbtable”：“测试”， “用户”：“管理员”， “密码”：“密码”

从 MySQL 5 读取 DynamicFrame

df_mysql5 =glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql5_options)

有没有办法给出一个 where 子句并说只从测试表中选择前 100 行，说它有一个名为“id”的列，我想使用这个查询来获取：

select * from test where id

感谢任何帮助。谢谢！

【问题讨论】：

我今天找到了这个，可能也是你要找的东西***.com/questions/51388993/…

【参考方案1】：

我能够提供自定义查询的方式是创建一个 Spark DataFrame 并使用选项指定它： https://spark.apache.org/docs/2.3.0/sql-programming-guide.html#manually-specifying-options

然后使用所述类将该 DataFrame 转换为 DynamicFrame： https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-dynamic-frame.html

tmp_data_frame = spark.read.format("jbdc")
.option("url", jdbc_url)
.option("user", username)
.option("password", password)
.option("query", "select * from test where id<100")
.load()

dynamic_frame = DynamicFrame.fromDF(tmp_data_frame, glueContext)

【参考方案2】：

抱歉，我会发表评论，但我没有足够的声誉。我能够使 Guillermo AMS 提供的解决方案在 AWS Glue 中工作，但它确实需要进行两项更改：

“jdbc”格式无法识别（提供的错误为：“py4j.protocol.Py4JJavaError：调用 o79.load 时出错。 ：java.lang.ClassNotFoundException：找不到数据源：jbdc。请在http://spark.apache.org/third-party-projects.html" 找到包）——我必须使用全名：“org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider” 查询选项对我不起作用（提供的错误是：“py4j.protocol.Py4JJavaError：调用 o72.load 时发生错误。 : java.sql.SQLSyntaxErrorException: ORA-00911: invalid character")，但幸运的是，“dbtable”选项支持传入表或子查询——即在查询周围使用括号。

在下面的解决方案中，我还围绕所需的对象和导入添加了一些上下文。 我的解决方案最终看起来像：

from awsglue.context import GlueContext
from pyspark.context import SparkContext

glue_context = GlueContext(SparkContext.getOrCreate())

tmp_data_frame = glue_context.spark_session.read\
  .format("org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider")\
  .option("url", jdbc_url)\
  .option("user", username)\
  .option("password", password)\
  .option("dbtable", "(select * from test where id<100)")\
  .load()