从 pyspark 中的另一个数据库加载 table

Question

我目前正在使用 AWS 和 PySpark。我的 table 存储在 S3 中，可从 Athena 查询。

在我的 Glue 作业中，我习惯于将我的 table 加载为：

my_table_df = sparkSession.table("myTable")

但是，这一次，我想从同一数据源 (AwsDataCatalog) 中的另一个数据库访问 table。所以我做了一些效果很好的事情：

my_other_table_df = sparkSession.sql("SELECT * FROM anotherDatabase.myOtherTable")

我只是在寻找一种更好的方法来编写相同的东西，而不使用 SQL 查询，在一行中，只需为此操作指定数据库。应该看起来像的东西

sparkSession.database("anotherDatabase").table("myOtherTable")

欢迎任何建议

Answer 1

您可以为此使用 DynamicFrameReader。这将 return 你一个 DynamicFrame。不过，您可以只对该 DynamicFrame 调用 .toDF() 以将其转换为本机 Spark DataFrame。

sc = SparkContext()
glue_context = GlueContext(sc)
spark = glue_context.spark_session
job = Job(glue_context)

data_source = glue_context.create_dynamic_frame.from_catalog(
            database="database",
            table_name="table_name"
).toDF()