Azure Databricks原有正常代码突发OutOfMemoryError求助

我在Azure Databricks中有一个Notebook，负责将bronze层表转换后写入silver层表，实现现有系统到新数据湖的初始加载，核心逻辑是将若干列转换为另一表的键值对子行。此前数周代码、数据、集群配置均未变更且运行正常，但近期执行时持续触发java.lang.OutOfMemoryError: Java heap space错误。目标表仅6列（3个整型、3个字符串型），数据量并不大。我已尝试分批处理、拆分Upsert调用、调整配置、更换更大规格服务器等方法，均无法解决问题。

原代码 #

dfCallData = fetchCallSummaryData(useArchive=False) \
                .join(_callMap, on=(col('UniqueCallID') == _callMap.systemCallId) & (_callMap.systemType != lit('NOVA')), how='inner') \
                .select(col('UniqueCallId'), col('tenantId'), col('agentId'), col('channelType'), col('callerPhone'), col('callerName'), col('terminationType'), \
                        col('callDuration'), col('transferDuration'), col('callOutcome'), col('transferReason'), col('endEvent'), col('lastStepName'), col('reportDate'), \
                        col('callInitiatedOn'), col('lastSystemUpdateOn'), col('isFirstTimeCaller'), col('isAfterHours'), col('finalStatus'), col('clk.isArchived'), \
                        col('clk.archivedOn'), col('Conversation'), col('QueueDuration'), col('AgentDuration'), col('TransferTalkDuration'), col('WhisperDuration'), \
                        col('BillingRatePerIVRMinute'), col('BillingRatePerTransferMinute'), col('BillingRatePerCall'), col('BillingGracePeriod'), col('CustomerData'), \
                        col('CustomerDataFlattened'), col('RunMode'), col('SurveyResults'), col('RedactedConversation'), col('Disposition'), col('Col1'), col('Col2'), \
                        col('Col3'), col('Col4'), col('Col5'), col('Col6'), col('Col7'), col('Col8'), col('Pub1'), col('Pub2'), col('Pub3'), col('Pub4'), col('Pub5'), \
                        col('Pub6'), col('Pub7'), col('Pub8'), col('PubBreadCrumbs'), col('PubFlattened'), col('PubLastBreadCrumb'), col('PubIntent'), \
                        col('PubAuthenticated'), col('Transcript'), col('clk.callId'))

pub_cols = ['Pub1', 'Pub2', 'Pub3', 'Pub4', 'Pub5', 'Pub6', 'Pub7', 'Pub8']

# Original method signature
def processMappedCallData(columns_to_convert:list) -> DataFrame:
    dfNewMap = spark.read.table('silver.appdata.call_data_map') \
                         .where(col('mapType') == 'columns') \
                         .alias('dm')

    dfCallDataSub = dfCallData.select('callId', 'UniqueCallId', col('agentId'), *columns_to_convert) \
                              .alias('cd')

    dfData = None
    for c in columns_to_convert:
        df = dfCallDataSub.join(dfNewMap, (col('cd.agentId') == col('dm.agentId')) & (col('dm.mapKey') == c), 'inner') \
                          .where((col(f'cd.{c}').isNotNull()) & (col(f'cd.{c}') != '')) \
                          .withColumn('callDataId', lit(None)) \
                          .withColumn('callDataType', lit('columns:mapped')) \
                          .select('callDataId', 'cd.callId', 'dm.callDataMapId', 'callDataType', lit(c).alias('legacyColumn'),  \
                                  col(f'cd.{c}').alias('dataValue'))

        dfData = dfData.union(df) if dfData is not None else df

    return dfData

dfPubCols = processMappedCallData(pub_cols)
_pipeline.execute_call_data_pipeline(dfPubCols, callDataType='columns')


# Upsert
def execute_call_data_pipeline(self, dfMappedData:DataFrame, callDataType='columns:mapped'):
        dtCallData = DeltaTable.forName(self._spark, f'{self.get_catalog()}.{self.get_schema()}.call_data')
        dtCallData.alias('old').merge(
                source=dfMappedData.alias('new'),
                condition=expr('old.callDataId = new.callDataId')
        ).whenMatchedUpdate(set=
            {
                'callId': col('new.callId') if 'callId' in dfMappedData.columns else col('old.callId'),
                'callDataMapId': col('new.callDataMapId') if 'callDataMapId' in dfMappedData.columns else col('old.callDataMapId'),
                'callDataType': col('new.callDataType') if 'callDataType' in dfMappedData.columns else col('old.callDataType'),
                'legacyColumn': col('new.legacyColumn') if 'legacyColumn' in dfMappedData.columns else col('old.legacyColumn'),
                'dataValue': col('new.dataValue') if 'dataValue' in dfMappedData.columns else col('old.dataValue'),
                'isEncrypted': col('new.isEncrypted') if 'isEncrypted' in dfMappedData.columns else col('old.isEncrypted'),
                'silverUpdateOn': lit(datetime.now(timezone.utc).timestamp())
            }
        ).whenNotMatchedInsert(values=
            {
                'callId': col('new.callId'),
                'callDataMapId': col('new.callDataMapId') if 'callDataMapId' in dfMappedData.columns else lit(None),
                'callDataType': col('new.callDataType') if 'callDataType' in dfMappedData.columns else lit(callDataType),
                'legacyColumn': col('new.legacyColumn') if 'legacyColumn' in dfMappedData.columns else lit(None),
                'dataValue': col('new.dataValue'),
                'isEncrypted': col('new.isEncrypted') if 'isEncrypted' in dfMappedData.columns else lit(False),
                'silverCreateOn': lit(datetime.now(timezone.utc).timestamp())
            }
        ).execute()

修改后的分批处理代码 #

def processMappedCallData(columns_to_convert:list):
    dfNewMap = spark.read.table('silver.appdata.call_data_map') \
                         .where(col('mapType') == 'columns') \
                         .alias('dm')

    dfCallDataSub = dfCallData.select('callId', 'UniqueCallId', col('agentId'), *columns_to_convert) \
                              .alias('cd')

    dfData = None
    for c in columns_to_convert:
        df = dfCallDataSub.join(dfNewMap, (col('cd.agentId') == col('dm.agentId')) & (col('dm.mapKey') == c), 'inner') \
                          .where((col(f'cd.{c}').isNotNull()) & (col(f'cd.{c}') != '')) \
                          .withColumn('callDataId', lit(None)) \
                          .withColumn('callDataType', lit('columns:mapped')) \
                          .select('callDataId', 'cd.callId', 'dm.callDataMapId', 'callDataType', lit(c).alias('legacyColumn'),  \
                                  col(f'cd.{c}').alias('dataValue'))
        _pipeline.execute_call_data_pipeline(df, callDataType='columns')

一、数据层面排查 #

• 检查源数据膨胀：即使目标表数据量小，fetchCallSummaryData返回的bronze层数据可能近期出现量级突增，或包含超大字段（比如Conversation、Transcript等文本字段存储了异常长的内容），导致内存占用飙升。可以通过dfCallData.count()、dfCallData.select(length('Conversation').alias('len')).describe()等语句统计数据规模和字段长度。
• 检查call_data_map表数据：该表可能新增了大量匹配记录，导致join后数据量远超预期。执行spark.read.table('silver.appdata.call_data_map').where(col('mapType') == 'columns').count()查看行数，对比之前的正常数值。

二、代码优化方向 #

• 替换循环Union为unpivot操作：原代码通过循环+Union拼接多个列的结果，会不断累加DataFrame的lineage，导致Spark执行计划过于复杂，内存压力增大。改用unpivot一次性处理所有列：

  # 示例unpivot写法
  dfCallDataSub = dfCallData.select('callId', 'UniqueCallId', 'agentId', *pub_cols)
  # 将指定列转为key-value形式
  dfUnpivoted = dfCallDataSub.selectExpr('callId', 'UniqueCallId', 'agentId', 
                                         'stack(8, "Pub1", Pub1, "Pub2", Pub2, "Pub3", Pub3, "Pub4", Pub4, "Pub5", Pub5, "Pub6", Pub6, "Pub7", Pub7, "Pub8", Pub8) as (legacyColumn, dataValue)') \
                             .where(col('dataValue').isNotNull() & (col('dataValue') != ''))
  # 再与call_data_map关联
  dfResult = dfUnpivoted.join(dfNewMap, (col('agentId') == dfNewMap.agentId) & (dfNewMap.mapKey == col('legacyColumn')), 'inner') \
                        .withColumn('callDataId', lit(None)) \
                        .withColumn('callDataType', lit('columns:mapped')) \
                        .select('callDataId', 'callId', dfNewMap.callDataMapId, 'callDataType', 'legacyColumn', 'dataValue')
  

• 缓存重复使用的大DataFrame：原代码中dfCallDataSub在循环中被多次join，每次都会重新计算。可以先对dfCallDataSub执行dfCallDataSub.cache()，用完后调用dfCallDataSub.unpersist()释放内存，减少重复计算的开销。
• 优化Merge逻辑：当前Merge条件是old.callDataId = new.callDataId，但新数据的callDataId是lit(None)，意味着所有新数据都会触发Insert操作，若数据量较大则会占用大量内存。可以：
  • 提前为新数据生成唯一callDataId，避免无效的匹配判断；
  • 按agentId或callId对Insert数据进一步分批次处理，缩小每批次数据量。

三、Spark配置与集群调优 #

• 调整内存参数：在集群配置中增加spark.executor.memory和spark.driver.memory，同时设置spark.executor.memoryOverhead为Executor内存的10%-20%，避免堆外内存不足引发OOM。
• 优化数据分区：检查dfCallData的分区数，若分区过少导致单分区数据量过大，可通过repartition(n)增加分区数，让数据均匀分布在多个Executor上。
• 禁用自动广播大表：如果call_data_map表数据量较大，Spark默认的广播join会将整个表加载到Driver内存导致OOM。可设置spark.sql.autoBroadcastJoinThreshold=-1禁用自动广播，或手动指定join策略为shuffle join。

四、运行时排查 #

• 分析Spark UI：在Databricks的Job页面打开Spark UI，定位OOM发生的具体阶段（join、Union或Merge），查看每个Task的内存使用和数据倾斜情况（比如某个Task处理的数据量远超其他Task）。
• 检查GC日志：开启Executor的GC日志，分析是否存在内存泄漏或频繁Full GC的情况，定位内存占用过高的根源。

内容的提问来源于stack exchange，提问作者Shane McGarry