amazon-web-services - 为什么我的 awsglue 作业仅使用一个执行程序和驱动程序？

Question

在我的脚本中，我将 pyspark 中的所有 dynamicframe 转换为 dataframe，并执行 groupby 和 join手术。然后在matrics View 中，我发现无论我设置多少DPU，只有一个执行器处于事件状态。

作业在大约 2 小时后失败，

Diagnostics: Container [pid=8417,containerID=container_1532458272694_0001_01_000001] is running beyond physical memory limits. Current usage: 5.5 GB of 5.5 GB physical memory used; 7.7 GB of 27.5 GB virtual memory used. Killing container.

我有大约 20 亿行数据。我的 DPU 设置为 80。

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "db", table_name = "in_json", transformation_ctx = "datasource0")
datasource1 = glueContext.create_dynamic_frame.from_catalog(database = "db", table_name = "out_json", transformation_ctx = "datasource0")


applymapping0 = ApplyMapping.apply(frame = datasource0, mappings = [("fieldA", "int", "fieldA", "int"), ("fieldB", "string", "fieldB", "string")], transformation_ctx = "applymapping1")
applymapping1 = ApplyMapping.apply(frame = datasource1, mappings = [("fieldA", "int", "fieldA", "int"), ("fieldB", "string", "fieldB", "string")], transformation_ctx = "applymapping1")

df1 = applymapping0.toDF().groupBy("fieldA"）.agg(count('*').alias("total_number_1"))
df2 = applymapping1.toDF().groupBy("fieldA"）.agg(count('*').alias("total_number_2"))

df1.join(df2, "fieldB")

result = DynamicFrame.fromDF(result_joined, glueContext, "result")

datasink2 = glueContext.write_dynamic_frame.from_options(frame = result, connection_type = "s3", connection_options = {"path": "s3://test-bucket"}, format = "json", transformation_ctx = "datasink2")
job.commit()

我错过了什么吗？

Answer 1

尝试重新分区您的DataFrame。可以重新分区based on a column ，或to an arbitrary number of partitions或both .

类似这样的事情:

df1 = applymapping0.toDF().groupBy("fieldA"）.agg(count('*').alias("total_number_1"))
df2 = applymapping1.toDF().groupBy("fieldA"）.agg(count('*').alias("total_number_2"))

df1_r = df1.repartition(df1("fieldB"))
df2_r = df2.repartition(df2("fieldB"))

df1_r.join(df2_r, "fieldB")