Facebook广告数据同步BigQuery异常：预览无数据及字段受限排查

开发了从Facebook广告管理后台拉取数据的函数，已成功生成BigQuery表结构，但BigQuery「PREVIEW」标签页显示「没有数据可展示」。仅使用精简字段请求时能正常获取数据，添加完整字段后无法写入数据，需排查代码问题。

逐一梳理代码中的核心问题：

1. 必填字段缺失（frequency） #

BigQuery schema中frequency字段标记为REQUIRED，但代码中既没有从Facebook Insights拉取该字段，也未设置默认值，导致数据插入时因必填字段为空被BigQuery拒绝。

修复：

• 在Facebook拉取字段中添加AdsInsights.Field.frequency
• 或将schema中frequency的mode改为NULLABLE，或在数据组装时给默认值0

2. 字段类型不匹配 #

Facebook返回的cpm、ctr、spend等字段是字符串格式（部分带货币符号、百分号），但BigQuery schema定义为NUMERIC，直接插入会触发类型转换错误。

修复：
新增类型转换函数，去除非数字字符后转为数值：

def convert_to_numeric(value):
    if not value:
        return 0
    # 去除$、%、逗号等非数字字符
    cleaned = ''.join(c for c in str(value) if c.isdigit() or c == '.')
    return float(cleaned) if cleaned else 0

3. 插入错误未捕获与日志缺失 #

insert_rows_bq函数中，client.insert_rows_json返回的错误信息resp未被处理，即使插入失败，日志仍显示「Success uploaded」，无法定位问题。

修复：
修改insert_rows_bq函数，添加错误判断与日志：

def insert_rows_bq(client, table_id, dataset_id, project_id, data):
    table_ref = "{}.{}.{}".format(project_id, dataset_id, table_id)
    table = client.get_table(table_ref)

    resp = client.insert_rows_json(
        json_rows=data,
        table=table_ref,
    )

    if resp:
        logger.error(f"Failed to upload to table {table.table_id}: {resp}")
        raise Exception(f"BigQuery insert error: {resp}")
    else:
        logger.info(f"Success uploaded {len(data)} rows to table {table.table_id}")

4. 日期处理逻辑错误 #

当event['attributes']中存在date时，代码直接调用strftime，但event的属性值通常是字符串，不是datetime对象，会触发AttributeError，导致函数中断，数据未插入。

修复：
统一日期处理逻辑，将字符串转为date对象：

if 'date' in event['attributes']:
    date_str = event['attributes']['date']
    yesterday = datetime.strptime(date_str, '%Y-%m-%d').date()
else:
    yesterday = date.today() - timedelta(1)

日期处理部分 #

def get_facebook_data(event, context):
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    bigquery_client = bigquery.Client()

    # 修复日期处理逻辑
    if 'date' in event['attributes']:
        date_str = event['attributes']['date']
        yesterday = datetime.strptime(date_str, '%Y-%m-%d').date()
    else:
        yesterday = date.today() - timedelta(1)
    # ... 后续代码

字段类型转换与数据组装部分 #

def convert_to_numeric(value):
    if not value:
        return 0
    cleaned = ''.join(c for c in str(value) if c.isdigit() or c == '.')
    return float(cleaned) if cleaned else 0

# ... 在组装fb_source时：
fb_source.append({
    'date': item['date_start'],
    'ad_id': item['ad_id'],
    'ad_name': item['ad_name'],
    'adset_id': item['adset_id'],
    'adset_name': item['adset_name'],
    'campaign_id': item['campaign_id'],
    'campaign_name': item['campaign_name'],
    'clicks': convert_to_numeric(item.get('clicks', 0)),
    'impressions': convert_to_numeric(item.get('impressions', 0)),
    'reach': convert_to_numeric(item.get('reach', 0)),
    'spend': convert_to_numeric(item.get('spend', 0)),
    'conversions': conversions,
    'actions': actions,
    'cpm': convert_to_numeric(item.get('cpm', 0)),
    'ctr': convert_to_numeric(item.get('ctr', 0)),
    'frequency': convert_to_numeric(item.get('frequency', 0)),  # 添加frequency字段处理
    'cost_per_action_type': cost_per_action_type
})

拉取字段补充 #

在account.get_insights的fields列表中添加AdsInsights.Field.frequency：

insights = account.get_insights(fields=[
        AdsInsights.Field.account_id,
        AdsInsights.Field.campaign_id,
        AdsInsights.Field.campaign_name,
        AdsInsights.Field.adset_name,
        AdsInsights.Field.adset_id,
        AdsInsights.Field.ad_name,
        AdsInsights.Field.ad_id,
        AdsInsights.Field.spend,
        AdsInsights.Field.impressions,
        AdsInsights.Field.reach,
        AdsInsights.Field.clicks,
        AdsInsights.Field.actions,
        AdsInsights.Field.conversions,
        AdsInsights.Field.cpm,
        AdsInsights.Field.ctr,
        AdsInsights.Field.cost_per_action_type,
        AdsInsights.Field.frequency  # 补充frequency字段
], params={
    'level': 'ad',
    'time_range': {
        'since':  yesterday.strftime("%Y-%m-%d"),
        'until': yesterday.strftime("%Y-%m-%d")
    },'time_increment': 1
})

1. 运行修改后的函数，查看日志是否有错误信息
2. 检查BigQuery表的Job History，查看插入任务的状态与错误详情
3. 用SELECT * FROM your_table WHERE date = 'yyyy-mm-dd'查询，确认数据是否写入

内容的提问来源于stack exchange，提问作者Artur Kurenov