如何在BigQuery sqlx脚本加载数据前验证云存储桶中CSV是否存在

要避免因GCS文件缺失导致脚本执行中断，可以通过先查询GCS存储桶元数据验证文件存在性，再条件执行LOAD DATA操作。以下是修改后的.sqlx脚本：

config {
    type: "operations",
    hasOutput: true
}

js {
    function getYesterday() {
        const today = new Date();
        const localOffset = -(today.getTimezoneOffset() / 60);
        const peruOffset = -5; // Peru timezone offset (GMT-5)
        const offset = peruOffset - localOffset;

        const yesterday = new Date(today.getTime() - 24 * 3600 * 1000 + offset * 3600 * 1000);

        const dd = String(yesterday.getDate()).padStart(2, '0');
        const mm = String(yesterday.getMonth() + 1).padStart(2, '0');
        const yyyy = yesterday.getFullYear();

        return dd + mm + yyyy;
    }
}

DECLARE file_exists BOOL;
DECLARE target_uri STRING;

SET target_uri = 'gs://bucket/test${getYesterday()}.csv';

-- 查询GCS存储桶元数据，验证文件是否存在
SET file_exists = EXISTS (
    SELECT 1
    FROM `bigquery-public-data.cloud_storage_geo_index.storage_objects`
    WHERE bucket_name = 'bucket'
      AND name = 'test${getYesterday()}.csv'
);

-- 仅当文件存在时执行数据加载
IF file_exists THEN
    LOAD DATA INTO
        project.dataset.incidents (incidentId Integer,
          name String,
          timestamp Integer)
    FROM FILES (skip_leading_rows=1, format = 'CSV',
        uris = [target_uri]);
ELSE
    SELECT '目标CSV文件不存在，跳过加载操作。' AS status;
END IF;

关键说明 #

• 元数据查询逻辑：利用BigQuery公开的cloud_storage_geo_index.storage_objects视图，通过匹配存储桶名称和文件名，判断目标文件是否存在。注意替换脚本中的bucket为你的实际存储桶名称。
• 条件执行控制：通过DECLARE定义变量存储目标URI和文件存在性结果，再用IF...ELSE分支控制LOAD DATA的执行。如果文件不存在，脚本会返回状态提示而非抛出错误中断执行。
• 时区兼容：保留了你原有的秘鲁时区日期计算逻辑，确保生成的文件名符合前一日的预期。

替代方案（无法访问公开视图时） #

如果你的环境无法使用cloud_storage_geo_index视图，可以在JS块中调用GCS的HEAD接口检查文件存在性，再将结果传入SQL逻辑：

config {
    type: "operations",
    hasOutput: true
}

js {
    function getYesterday() {
        const today = new Date();
        const localOffset = -(today.getTimezoneOffset() / 60);
        const peruOffset = -5;
        const offset = peruOffset - localOffset;

        const yesterday = new Date(today.getTime() - 24 * 3600 * 1000 + offset * 3600 * 1000);

        const dd = String(yesterday.getDate()).padStart(2, '0');
        const mm = String(yesterday.getMonth() + 1).padStart(2, '0');
        const yyyy = yesterday.getFullYear();

        return dd + mm + yyyy;
    }

    async function checkFileExists() {
        const fileName = `test${getYesterday()}.csv`;
        const bucketName = 'bucket';
        const url = `https://storage.googleapis.com/storage/v1/b/${bucketName}/o/${encodeURIComponent(fileName)}`;
        
        try {
            const response = await fetch(url, { method: 'HEAD' });
            return response.ok;
        } catch (err) {
            return false;
        }
    }
}

DECLARE file_exists BOOL DEFAULT ${checkFileExists()};
DECLARE target_uri STRING DEFAULT 'gs://bucket/test${getYesterday()}.csv';

IF file_exists THEN
    LOAD DATA INTO
        project.dataset.incidents (incidentId Integer,
          name String,
          timestamp Integer)
    FROM FILES (skip_leading_rows=1, format = 'CSV',
        uris = [target_uri]);
ELSE
    SELECT '目标CSV文件不存在，跳过加载操作。' AS status;
END IF;

注意：使用该方案需确保BigQuery脚本的执行账号拥有访问目标GCS存储桶的权限（通过IAM角色或服务账号配置）。

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