我有一个代码,它将两个orc文件读取为两个数据帧,并将它们连接到一个数据帧中。我的代码,然后将此数据帧写入文件。我试图知道每一步的时间,不写输出,它需要一分钟,但当我写代码,它需要大约38分钟。数据为5gb,1亿行50列。你知道吗
代码是:
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 pyspark.sql import DataFrameReader, DataFrameWriter
from datetime import datetime
import time
# @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
print("All imports were successful.")
df = spark.read.orc(
's3://****'
)
print("First dataframe read with headers set to True")
df2 = spark.read.orc(
's3://****'
)
print("Second dataframe read with headers set to True")
# Obtain columns lists
left_cols = df.columns
right_cols = df2.columns
# Prefix each dataframe's field with "left_" or "right_"
df = df.selectExpr([col + ' as left_' + col for col in left_cols])
df2 = df2.selectExpr([col + ' as right_' + col for col in right_cols])
# Perform join
# df3 = df.alias('l').join(df2.alias('r'), on='l.left_c_0' == 'r.right_c_0')
# df3 = df.alias('l').join(df2.alias('r'), on='c_0')
df3 = df.join(
df2,
df["left_c_0"] == df2["right_c_0"]
)
print("Dataframes have been joined successfully.")
output_file_path = 's3://****'.format(
datetime.utcnow()
)
df3.write.orc(
output_file_path
)
# print("Dataframe has been written to csv.")
job.commit()
为了清楚起见,当我注释掉df3.write.orc( output_file_path )
时,我可以在1分钟内完成脚本的工作。你知道吗
我想不出什么方法能更快地完成写作任务。这里有什么可以并行化的吗?你知道吗
注意,输出文件是部分编写的。你知道吗
当您创建数据帧时,它不会从s3读取。数据帧是关于从何处(以及如何)读取数据的信息。这同样适用于连接数据帧时,它只是创建了一个新的数据帧,该数据帧知道应该处理前两个数据帧以获得结果。你知道吗
最后,当您调用
df3.write
时,即读取和处理df1
和df2
的数据。实际上,所有的应用程序处理(从s3读取数据、重命名列、连接两个数据帧)都发生在df3.write
的调用过程中。这就是为什么如果删除最后一条语句
df3.write
(因为实际上什么都没做),脚本会在不到1秒的时间内完成。你知道吗真正的问题是如何让这个应用程序更快。它取决于许多因素,如输入数据大小、内存和可用于该作业的处理核心的数量。你知道吗
相关问题 更多 >
编程相关推荐