AWS胶PySpark替换空值

Question

我正在运行一个AWS Glue作业，使用从Glue自动生成的PySpark脚本，将S3上的管道分隔文件加载到RDS Postgres实例中。

最初，它抱怨某些列中的空值：

pyspark.sql.utils.IllegalArgumentException: u"Can't get JDBC type for null"

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = 
"xyz_catalog", table_name = "xyz_staging_files", transformation_ctx = 
"datasource0")
custom_df = datasource0.toDF()
custom_df2 = custom_df.fillna(-1)
custom_df3 = custom_df2.fromDF()

applymapping1 = ApplyMapping.apply(frame = custom_df3, mappings = [("id", 
"string", "id", "int"),........more code

http://spark.apache.org/docs/latest/api/python/pyspark.sql.sql.html#pyspark.sql.dataframe.fillna

现在，当我运行作业时，它会抛出以下错误：

Log Contents:
Traceback (most recent call last):
File "script_2017-12-20-22-02-13.py", line 23, in <module>
custom_df3 = custom_df2.fromDF()
AttributeError: 'DataFrame' object has no attribute 'fromDF'
End of LogType:stdout

我对Python和Spark是新手，尝试了很多，但不明白这一点。感谢一些专家在这方面的帮助。

custom_df3 = glueContext.create_dynamic_frame.fromDF(frame = custom_df2)

AttributeError: 'DynamicFrameReader' object has no attribute 'fromDF'

pyspark.sql.utils.IllegalArgumentException: u"Can't get JDBC type for null"

更新：确保DynamicFrame已经导入（从awsglue.context import DynamicFrame)，因为fromDF/toDF是DynamicFrame的一部分。

请参阅https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-dynamic-frame.html

Answer 1

你在打电话。来自错误的班级。应该是这样的：

from awsglue.dynamicframe import DynamicFrame
DyamicFrame.fromDF(custom_df2, glueContext, 'label')