所以我使用 AWS Glue 自动生成的代码从 S3 读取 csv 文件并通过 JDBC 连接将其写入表。看起来很简单,Job 成功运行,没有错误,但它什么也没写。当我检查 Glue Spark 动态框架时,它包含所有行(使用 .count())。但是什么时候做 .show() 什么都不会产生。
.printSchema() 工作正常。尝试在使用 .show() 时记录错误,但没有错误或未打印任何内容。使用 .toDF 及其工作的显示方法将 DynamicFrame 转换为数据框。 我认为文件有问题,试图缩小到某些列。但即使文件中只有 2 列也是一样的。明明用双引号标明了字符串,还是没有成功
我们需要从 Glue 配置中选择 JDBC 连接之类的东西。我想这是普通的 spark 数据框做不到的。因此需要动态框架工作。
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
import logging
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
glueContext = GlueContext(SparkContext.getOrCreate())
spark = glueContext.spark_session
datasource0 = glueContext.create_dynamic_frame.from_options('s3', {'paths': ['s3://bucket/file.csv']}, 'csv', format_options={'withHeader': True,'skipFirst': True,'quoteChar':'"','escaper':'\\'})
datasource0.printSchema()
datasource0.show(5)
输出
root
|-- ORDERID: string
|-- EVENTTIMEUTC: string
这是转换为常规数据框的结果。
datasource0.toDF().show()
输出
+-------+-----------------+
|ORDERID| EVENTTIMEUTC|
+-------+-----------------+
| 2| "1/13/2018 7:50"|
| 3| "1/13/2018 7:50"|
| 4| "1/13/2018 7:50"|
| 5| "1/13/2018 7:50"|
| 6| "1/13/2018 8:52"|
| 7| "1/13/2018 8:52"|
| 8| "1/13/2018 8:53"|
| 9| "1/13/2018 8:53"|
| 10| "1/16/2018 1:33"|
| 11| "1/16/2018 2:28"|
| 12| "1/16/2018 2:37"|
| 13| "1/17/2018 1:17"|
| 14| "1/17/2018 2:23"|
| 15| "1/17/2018 4:33"|
| 16| "1/17/2018 6:28"|
| 17| "1/17/2018 6:28"|
| 18| "1/17/2018 6:36"|
| 19| "1/17/2018 6:38"|
| 20| "1/17/2018 7:26"|
| 21| "1/17/2018 7:28"|
+-------+-----------------+
only showing top 20 rows
这是一些数据。
ORDERID, EVENTTIMEUTC
1, "1/13/2018 7:10"
2, "1/13/2018 7:50"
3, "1/13/2018 7:50"
4, "1/13/2018 7:50"
5, "1/13/2018 7:50"
6, "1/13/2018 8:52"
7, "1/13/2018 8:52"
8, "1/13/2018 8:53"
9, "1/13/2018 8:53"
10, "1/16/2018 1:33"
11, "1/16/2018 2:28"
12, "1/16/2018 2:37"
13, "1/17/2018 1:17"
14, "1/17/2018 2:23"
15, "1/17/2018 4:33"
16, "1/17/2018 6:28"
17, "1/17/2018 6:28"
18, "1/17/2018 6:36"
19, "1/17/2018 6:38"
20, "1/17/2018 7:26"
21, "1/17/2018 7:28"
22, "1/17/2018 7:29"
23, "1/17/2018 7:46"
24, "1/17/2018 7:51"
25, "1/18/2018 2:22"
26, "1/18/2018 5:48"
27, "1/18/2018 5:50"
28, "1/18/2018 5:50"
29, "1/18/2018 5:51"
30, "1/18/2018 5:53"
100, "1/18/2018 10:32"
101, "1/18/2018 10:33"
102, "1/18/2018 10:33"
103, "1/18/2018 10:42"
104, "1/18/2018 10:59"
105, "1/18/2018 11:16"
阅读 DynamicFrame
的
docs,他们对此不是很明确,但听起来可能存在没有处理过的 underyling
DataFrame
的情况,直到您调用 toDF()
,所以您实际上是在调用 .show()
在某物上 可以是空白的:
为了解决这些限制,AWS Glue 引入了 DynamicFrame。 A DynamicFrame 类似于 DataFrame,除了每条记录是 自描述,因此最初不需要模式。相反,AWS Glue 在需要时即时计算模式,并显式编码 使用选择(或联合)类型的模式不一致。你可以解决 这些不一致使您的数据集与数据兼容 需要固定模式的商店。
.toDF():
通过转换将 DynamicFrame 转换为 Apache Spark DataFrame DynamicRecords 到 DataFrame 字段。返回新的 DataFrame。
在此处检查代码,似乎存在底层 Java 数据框为空白的实例:https://github.com/awslabs/aws-glue-libs/blob/f973095b9f2aa784cbcc87681a00da3127125337/awsglue/dynamicframe.py#L78
def show(self, num_rows=20):
print(self._jdf.showString(num_rows))
其中
__init__
依赖于传递的参数 (jdf
) 可能尚未收集:
def __init__(self, jdf, glue_ctx, name=""):
self._jdf = jdf
self.glue_ctx = glue_ctx
self._ssql_ctx = glue_ctx._ssql_ctx
self._sc = glue_ctx and glue_ctx._sc
self._schema = None
self._lazy_rdd = None
self.name = name
当调用
toDF()
底层数据帧被处理:
... SNIP ...
scala_options.append(self.glue_ctx.convert_resolve_option(option.path, option.action, option.target))
return DataFrame(self._jdf.toDF(self.glue_ctx._jvm.PythonUtils.toSeq(scala_options)), self.glue_ctx)
.toDF()
的
Java 文档提到此方法从 RDD 转换(即它正在收集工人的结果):
这在从元组的 RDD 转换为
有意义的名字。DataFrame