以下是一个完整的 PySpark 解决方案用于根据输入的表名和过滤条件生成 INSERT 语句。代码考虑了不同数据类型的格式化、NULL 处理、单引号转义以及列名的安全引用并提供了两种使用方式收集到驱动端和分布式写入文件。函数实现frompyspark.sqlimportDataFramefrompyspark.sql.typesimport(StringType,IntegerType,LongType,ShortType,ByteType,FloatType,DoubleType,DecimalType,BooleanType,DateType,TimestampType)defgenerate_insert_statements(table_name:str,filter_condition:str,target_table:strNone,batch_size:int10000): 根据源表和过滤条件生成 INSERT 语句。 参数: table_name (str): 源表名可包含数据库名如 db.table。 filter_condition (str): 过滤条件SQL WHERE 子句不含 WHERE 关键字。 target_table (str, optional): 目标表名若不提供则使用源表名。 batch_size (int, optional): 分批处理大小用于控制内存占用仅在分布式写入时使用。 返回: list: 包含 INSERT 语句的列表当使用 collect 模式时。 或 None当使用分布式写入模式时结果直接写入文件。 iftarget_tableisNone:target_tabletable_name# 读取源表并应用过滤dfspark.table(table_name).filter(filter_condition)columnsdf.columns schemadf.schemadefformat_value(value,data_type):将 PySpark 值转换为 SQL 字面量字符串ifvalueisNone:returnNULL# 字符串类型单引号包裹内部单引号转义为两个单引号ifisinstance(data_type,StringType):returnstr(value).replace(,)# 数值类型直接转为字符串elifisinstance(data_type,(IntegerType,LongType,ShortType,ByteType,FloatType,DoubleType,DecimalType)):returnstr(value)# 布尔类型使用 TRUE/FALSE可根据目标数据库调整为 1/0elifisinstance(data_type,BooleanType):returnTRUEifvalueelseFALSE# 日期和时间戳转为字符串并加单引号ISO 格式elifisinstance(data_type,(DateType,TimestampType)):returnstr(value)# 其他类型如二进制、复杂类型默认按字符串处理并给出警告else:print(f警告未处理的类型{data_type}按字符串处理)returnstr(value).replace(,)# 构建列名列表使用反引号避免特殊字符问题column_list, .join([f{c}forcincolumns])# 使用迭代器逐行生成语句避免一次性加载所有数据到内存defrow_to_insert(row):values[]forcol_name,fieldinzip(columns,schema.fields):valrow[col_name]values.append(format_value(val,field.dataType))values_str, .join(values)returnfINSERT INTO {target_table} ({column_list}) VALUES ({values_str});# 方式一收集到驱动端返回列表适用于小数据量# return [row_to_insert(row) for row in df.collect()]# 方式二使用 toLocalIterator 生成器逐条返回可配合外部循环写入文件defgenerate():forrowindf.toLocalIterator():yieldrow_to_insert(row)returngenerate()使用示例# 生成 INSERT 语句返回生成器可迭代处理insert_gengenerate_insert_statements(table_namesales_db.orders,filter_conditionorder_date 2024-01-01 AND status completed,target_tablearchive_db.orders)# 示例打印前 5 条fori,stmtinenumerate(insert_gen):ifi5:breakprint(stmt)# 示例将所有语句写入文件withopen(/dbfs/tmp/insert_statements.sql,w)asf:forstmtingenerate_insert_statements(sales_db.orders,status completed):f.write(stmt\n)注意事项大数据量· 使用 collect() 会将所有数据加载到驱动节点可能导致内存溢出。· 推荐使用 toLocalIterator() 生成器逐条处理或使用分布式写入文件见下文“分布式生成方式”。复杂数据类型· 对于数组、结构体、映射等复杂类型上述代码会按字符串处理可能不符合目标数据库语法。· 建议针对具体类型扩展 format_value 函数如序列化为 JSON 或使用目标数据库支持的格式。SQL 方言差异· 布尔值 TRUE/FALSE 在 MySQL、PostgreSQL 中通用但 SQL Server 使用 1/0Oracle 使用 1/0 或 ‘Y’/‘N’。· 日期和时间戳格式可能需要调整如 Oracle 的 TO_DATE。转义与安全性· 字符串内部单引号已处理为两个单引号符合 SQL 标准。· 表名和列名使用反引号包裹避免与数据库关键字冲突。性能优化· 若数据量极大建议直接在 Spark 中构建 INSERT 语句字符串列然后使用 df.write.text() 分布式写出避免驱动程序成为瓶颈。分布式生成方式可选以下方法利用 Spark 的分布式能力直接在各个分区生成 INSERT 语句并写入文件适合大规模数据。frompyspark.sql.functionsimportcol,lit,concat,when,isnull,format_stringfrompyspark.sql.typesimportStringTypedefgenerate_insert_statements_distributed(table_name:str,filter_condition:str,target_table:strNone,output_path:str/dbfs/tmp/insert_statements):iftarget_tableisNone:target_tabletable_name dfspark.table(table_name).filter(filter_condition)columnsdf.columns# 为每个列构建 SQL 字面量表达式value_exprs[]forcincolumns:col_typedf.schema[c].dataType col_exprcol(c)ifisinstance(col_type,StringType):# 字符串添加单引号并转义exprconcat(lit(),col_expr.cast(StringType()),lit())exprexpr.replace(,)# 注意此方法在 Spark 中不可用需使用 regexp_replaceexprconcat(lit(),regexp_replace(col_expr.cast(StringType()),,),lit())elifisinstance(col_type,(IntegerType,LongType,ShortType,ByteType,FloatType,DoubleType,DecimalType)):exprcol_expr.cast(StringType())elifisinstance(col_type,BooleanType):exprwhen(col_expr,lit(TRUE)).otherwise(lit(FALSE))elifisinstance(col_type,(DateType,TimestampType)):exprconcat(lit(),col_expr.cast(StringType()),lit())else:exprconcat(lit(),col_expr.cast(StringType()),lit())# 处理 NULL使用 coalesce 将 NULL 替换为字符串 NULL不加引号exprwhen(isnull(col_expr),lit(NULL)).otherwise(expr)value_exprs.append(expr)# 构建完整的 INSERT 语句列column_list_str, .join([f{c}forcincolumns])values_exprconcat(lit((),concat_ws(, ,*value_exprs),lit()))insert_stmt_colconcat(lit(fINSERT INTO {target_table} ({column_list_str}) VALUES ),values_expr,lit(;))# 选择该列并写入文本文件每个分区一个文件df.select(insert_stmt_col.alias(insert_sql)).write.mode(overwrite).text(output_path)说明· 使用 regexp_replace 转义字符串中的单引号。· 使用 when(isnull(col), …) 区分 NULL 和其他值。· 结果通过 write.text() 分布写入指定路径每个分区生成一个文件避免驱动端内存压力。总结以上代码提供了从 PySpark 表生成 INSERT 语句的完整实现。根据数据量大小和具体需求可以选择简单模式收集到驱动端或分布式模式直接写入文件。使用时请根据目标数据库的语法调整数据类型格式化和转义规则。