AWS Glue优化IoT遥测数据转换：解决写入慢问题
2026-6-13
问题描述
我在S3桶中存储了2年的JSON格式IoT遥测数据，希望通过AWS Glue将其转换为按年、月、日、小时、产品类型、MAC地址分区的结构后，写入数据湖中的另一个S3桶。但当前使用Glue作业执行时写入速度极慢，预估耗时至少70小时，现寻求快速且低成本的解决方案，当前使用的PySpark代码如下：
import sys from awsglue.utils import getResolvedOptions import json from datetime import datetime import pyspark.sql.functions as F from pyspark.sql import SparkSession from pyspark.sql.functions import lit, col, explode, arrays_zip from pyspark.sql.types import (StructType, StructField, StringType, LongType, ArrayType) import logging logging.basicConfig(level=logging.INFO) import boto3 from pyspark.sql.functions import col, to_timestamp, year, month, dayofmonth, when, lit,trim ################ 从S3读取JSON数据和配置文件 开始 ####################### def get_configuration_file_from_s3(bucket, file_key, region_name): # 初始化boto3以使用S3客户端 s3_client = boto3.client('s3', region_name=region_name) # 获取S3中的文件 s3_response = s3_client.get_object( Bucket=bucket, Key=file_key ) # 从响应中获取Body对象并以字节格式读取数据 s3_obj_body = s3_response.get('Body') content = s3_obj_body.read() try: # 解析JSON json_dict = json.loads(content) except json.decoder.JSONDecodeError as e: logging.info(f'JSON file is not properly formatted. ERROR: {e}') return json_dict ################ 从S3读取JSON数据和配置文件 结束 ####################### ########################## 将输出数据写入S3 开始 ############################ def write_json(processed_df, output_s3_path, partition_columns, num_partitions=20): ''' 将分区后的JSON文件写入指定位置 :param processed_df: 待写入的输入Spark Dataframe :param output_s3_path: 输出文件路径的URI :param num_partitions: 分区列内的分区数量 ''' logging.info('write the data into s3 Started {}'.format(datetime.now())) processed_df.repartition(num_partitions, partition_columns).write.mode('append').option( 'maxRecordsPerFile', 9000000).partitionBy( partition_columns).format('json').save(output_s3_path) logging.info('writing completed {0} with -- {1}'.format(output_s3_path, datetime.now())) ########################## 将输出数据写入S3 结束 ############################ ########################## 根据process_timestamp生成原始路径 开始 ################## def raw_path(process_timestamp, source_bucket, input_path, data_format, config_bucket, data_format_config): process_date = datetime.strptime(process_timestamp[0:19], '%Y-%m-%d %H:%M:%S') process_date_month = f'{process_date.month}' if process_date.month >9 else f'0{process_date.month}' process_date_day = f'{process_date.day}' if process_date.day >9 else f'0{process_date.day}' read_path = f"{source_bucket}/{input_path}/history_{process_date.year}-{process_date_month}-{process_date_day}*.json" return read_path def process_column(df, column_name): field_type = df.schema[column_name].dataType if isinstance(field_type, ArrayType): #return col(column_name)[0].alias(column_name) return when(col(column_name).isNotNull(), col(column_name)[0]).otherwise(lit(None)).alias(column_name) elif isinstance(field_type, StructType): # 处理嵌套结构体，如{"$numberDouble": "3120.7"}或{"$numberInt": "0"} field_names = [f.name for f in field_type.fields] if "$numberDouble" in field_names: #return col(f"{column_name}.$numberDouble").cast("float").alias(column_name) return when(col(column_name).isNotNull(), col(f"{column_name}.$numberDouble").cast("float")).otherwise(lit(None)).alias(column_name) elif "$numberInt" in field_names: #return col(f"{column_name}.$numberInt").cast("int").alias(column_name) return when(col(column_name).isNotNull(), col(f"{column_name}.$numberInt").cast("int")).otherwise(lit(None)).alias(column_name) elif "$date" in field_names: #return col(f"{column_name}.$date.$numberLong").cast("long").alias(column_name) return when(col(column_name).isNotNull(), col(f"{column_name}.$numberLong").cast("long")).otherwise(lit(None)).alias(column_name) else: #return col(column_name) return when(col(column_name).isNotNull(),col(column_name)).otherwise(lit(None)).alias(column_name) elif isinstance(field_type, StringType): #return trim(col(column_name)).alias(column_name) return when((col(column_name).isNotNull()) & (trim(col(column_name)) != ''),col(column_name)).otherwise(lit(None)).alias(column_name) else: #return col(column_name).alias(column_name) return when((col(column_name).isNotNull()) & (trim(col(column_name)) != ''),col(column_name)).otherwise(lit(None)).alias(column_name) def run_spark_job(device_history_source, curated_output_path, config_bucket, data_format, mac_address_config, region_name): logging.info('Script Started {}'.format(datetime.now())) application_name = 'unit level data processing-test' spark = SparkSession.builder.appName(application_name) \ .config('spark.sql.broadcastTimeout', '36000') \ .config("spark.sql.caseSensitive",True)\ .config("spark.default.parallelism", "200") \ .config("spark.rpc.message.maxSize", "2047") \ .config("spark.shuffle.service.enabled ", "true") \ .config("spark.driver.memory", "40G") \ .config("spark.driver.memoryOverhead", "40G") \ .config("spark.executor.memory", "40G") \ .config("spark.executor.memoryOverhead", "40G") \ .config("spark.driver.maxResultSize", "40G") \ .config("spark.shuffle.spill.compress", "true") \ .config("spark.shuffle.compress", "true") \ .config("spark.sql.optimizer.dynamicPruning.enforceBroadcastReuse", "true") \ .config("spark.sql.optimizer.distinctBeforeIntersect.enabled", "true") \ .config("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true") \ .config("spark.sql.parquet.fs.optimized.committer.optimization-enabled", "false") \ .config("yarn.nodemanager.pmem-check-enabled", "true").getOrCreate() sc = spark.sparkContext try: mac_address_data = get_configuration_file_from_s3(config_bucket, mac_address_config, region_name) logging.info('Read the data from s3 Started {}'.format(datetime.now())) input_data = spark.read.json(device_history_source).dropDuplicates() logging.info('Read the data from s3 completed {}'.format(datetime.now())) mac_addresses = mac_address_data.get('mac_addresses', '*') # 允许的产品类型列表 allowed_product_types = [ 'heatpumpWaterHeaterGen1', 'heatpumpWaterHeaterGen4', 'heatpumpWaterHeaterGen5', ] # 定义过滤规则 logging.info('Process Fitring Started {}'.format(datetime.now())) ignore_fields = ['_id','device_base_address','created_at','transactionId', 'device_name', 'connected', 'cloud2_account_id', \ 'cloud2_instance_id', 'account_id', 'user_connected', 'location_id', 'user_id', \ 'name', 'software_version', 'deviceId', \ '@DATA_SHARING_PARTNER_ID', '@partner_id','@PERFORM_SCHEDULE'] #print("......Process Fitring completed count ",df_filtered.count()) except Exception as e: logging.error(f"Error occurred during processing: {str(e)}") return None logging.info('Process history_df Started {}'.format(datetime.now())) #df = df_filtered.withColumn("timestamp1", to_timestamp("timestamp"))#.drop(col("SCHEDULS")) df_filter = input_data.drop(*ignore_fields).filter((col("mac_address").isin(mac_addresses)) & (col("econetProductType").isin(allowed_product_types))) if 'product_type' in df_filter.columns: df_filter = df_filter.drop('product_type') df_filtered = df_filter.withColumnRenamed("econetProductType", "product_type") # 提取年、月、日 df = df_filtered.withColumn("year", F.substring("timestamp", 1, 4).cast("int")) \ .withColumn("month", F.substring("timestamp", 6, 2).cast("int")) \ .withColumn("day", F.substring("timestamp", 9, 2).cast("int")) logging.info('Process history_df completed {}'.format(datetime.now())) logging.info('Process Fitring completed {}'.format(datetime.now())) # 应用转换 logging.info('Process df_transformed Started {}'.format(datetime.now())) df_transformed = df.select( *[process_column(df, col_name) for col_name in df.columns] ) logging.info('Process df_transformed completed {}'.format(datetime.now())) logging.info('Process converted into string Started {}'.format(datetime.now())) history_final = df_transformed.select([col(c).cast("string").alias(c) for c in df.columns]) logging.info('Process converted into string completed {}'.format(datetime.now())) if history_final is not None: #history_final.persist() write_json(history_final, curated_output_path, ['year', 'month', 'day','mac_address', 'product_type']) logging.info('Data Storage Ends {}'.format(datetime.now())) logging.info('Script End {}'.format(datetime.now())) if __name__ == '__main__': args = getResolvedOptions( sys.argv, [ 'JOB_NAME', 'source_bucket', 'output_bucket', 'config_bucket', 'input_path', 'data_format', 'process_timestamp', 'mac_address_config', 'data_format_config', 'region_name' ] ) source_bucket = args['source_bucket'].strip() output_bucket = args['output_bucket'].strip() config_bucket = args['config_bucket'].strip() input_path = args['input_path'].strip() data_format = args['data_format'].strip() process_timestamp = args['process_timestamp'].strip() mac_address_config = args['mac_address_config'].strip() #mac_address_config = 'staging/config/mac_address_config_24622_18_March_24_March.json' data_format_config = args['data_format_config'].strip() region_name = args['region_name'].strip() device_history_source = raw_path(process_timestamp, source_bucket, input_path, data_format, config_bucket, data_format_config) logging.info(f"Processing {device_history_source}") device_history_output = output_bucket + f'/telemetry_data/' logging.info(f"Processing2 {device_history_source}") run_spark_job(device_history_source, device_history_output, config_bucket, data_format, mac_address_config, region_name)
解决方案
一、存储格式替换：JSON → Parquet
JSON为行式存储，读写效率远低于列式存储的Parquet，切换后可提升IO性能3-5倍，同时支持压缩减少存储成本：
- 将写入格式从
.format('json')改为.format('parquet') - 添加
.option('compression', 'snappy')启用Snappy压缩，兼顾压缩比与速度
二、Spark配置调优
清理冗余配置，优化并行与IO参数：
- 启用Parquet优化提交器：将
spark.sql.parquet.fs.optimized.committer.optimization-enabled设为true，减少S3写入的HTTP请求数 - 调整并行度：
spark.default.parallelism设为executor数量×executor核心数×2（如20个4核executor则设为160），避免任务并行度过低 - 移除无效配置：删除
yarn.nodemanager.pmem-check-enabled，Glue无需YARN节点内存校验；删除手动设置的spark.driver.memory等参数，改用Glue作业DPU配置分配资源
三、代码逻辑优化
1. 提前过滤数据
将dropDuplicates()移至过滤逻辑之后，减少去重数据量；读取时利用S3前缀分区批量读取，避免全量扫描：
# 先过滤再去重 df_filter = input_data.drop(*ignore_fields).filter((col("mac_address").isin(mac_addresses)) & (col("econetProductType").isin(allowed_product_types))).dropDuplicates()
2. 完善分区策略
- 补充
hour分区列：从timestamp提取小时字段，满足需求中的小时分区要求 - 调整分区列顺序为基数从高到低：
year, month, day, hour, product_type, mac_address，减少小文件生成
# 提取完整时间分区列 df = df_filtered.withColumn("year", F.substring("timestamp", 1, 4).cast("int")) \ .withColumn("month", F.substring("timestamp", 6, 2).cast("int")) \ .withColumn("day", F.substring("timestamp", 9, 2).cast("int")) \ .withColumn("hour", F.substring("timestamp", 12, 2).cast("int"))
3. 避免无意义类型转换
删除将所有列转为String类型的逻辑，仅保留嵌套结构到基础类型的转换，减少计算开销与存储空间浪费
4. 优化配置文件读取
用Spark分布式读取替换boto3单线程读取，提升配置加载效率：
# 用Spark读取配置文件 mac_address_df = spark.read.json(f"s3://{config_bucket}/{mac_address_config}") mac_addresses = mac_address_df.first().get('mac_addresses', '*')
5. 调整repartition策略
根据数据量设置合理分区数（按每个分区1GB计算），避免小文件过多：
# 按分区列+数据量设置分区数 processed_df.repartition(100, *partition_columns)
四、Glue作业配置优化
- 选择高性能DPU类型：使用G.2X（4vCPU/16GB）或G.4X（8vCPU/32GB）类型，相比标准型提升计算能力
- 启用自动缩放：开启Glue作业自动缩放，根据数据量动态调整executor数量
- 批量拆分任务：将2年数据按月份拆分，每月运行一次作业，避免单次处理数据量过大
- 启用作业书签：避免重复处理已完成的数据，提升增量处理效率
修改后的核心代码片段
# 优化的Parquet写入函数 def write_parquet(processed_df, output_s3_path, partition_columns, num_partitions=100): logging.info('write the data into s3 Started {}'.format(datetime.now())) processed_df.repartition(num_partitions, *partition_columns).write.mode('append') \ .option('maxRecordsPerFile', 1000000) \ .option('compression', 'snappy') \ .partitionBy(partition_columns) \ .format('parquet') \ .save(output_s3_path) logging.info('writing completed {0} with -- {1}'.format(output_s3_path, datetime.now())) # 调用写入函数时传入完整分区列 write_parquet(history_final, curated_output_path, ['year', 'month', 'day', 'hour', 'product_type', 'mac_address'])
内容的提问来源于stack exchange，提问作者Ramanarao
相关产品推荐
相关产品推荐
