时间戳

首页 > 解决方案 > 如何根据映射文件更新大查询数据？

python - 如何根据映射文件更新大查询数据？

问题描述

我有一个存储在 Gcs 中的映射文件和一个大查询表。

df_mapping：

             Id     Values
             1      XZUP
             2      SJXC
             3      PALD
             4      QLOM
             5      DKCM

BQ_表：

Id  Country  Market  Sales  Values
1   Canada   Hsp     2503   XZUP
2   Germany  Noe     2459   SJXC
3   Algeria  Zoe     4635   PALD
4   Brazil   Foe     6354   QLOM
5   Canada   Cmm     2588   XZUP

但是映射文件每次都会更改，如果映射文件更改我的 BQ 表值也应该根据映射文件更新，我正在寻找解决方案。

我做了什么：Everytime mapping file changes, i am triggering a function where i am reading bq table except "value" column & reading updated mapping file --> left join on Id column to get updated "values" --> Deleting my old bq table --> Inserting the new data.

query =  """
            SELECT
            Id,
            Country,
            Sales,
            Value
            FROM `project.dataset.tbl` 
            
        """

            bqclient = bigquery.Client()
            df = (
                bqclient.query(query)
                .result()
                .to_dataframe(create_bqstorage_client=True)
                )

df_mapping = pd.read_csv("gs://path/mapping.csv")
df_final = pd.merge(df, df_mapping, on='Id', how='left')
-- Not sure of Deleting and Inserting data safely

我不确定，删除我的旧表并插入新表是解决这个问题的正确方法。

我看到的问题：

          1. After deleting my old table, error can come while inserting new data.
          2. Data is quite large to process ~1million.
          3. Not Scalable solution.
          4. Can loss Data.

有没有其他方法可以做到这一点，比如在 BQ Table 中它引用我的映射文件或任何其他解决此任务的方法都会很棒。

标签： pythonpandasgoogle-bigquerygoogle-cloud-functionsgoogle-cloud-storage

解决方案

让我们试一试——

  • 在 BigQuery 中创建计划的 UPDATE 查询（“按需”）。在这里，您可以将 GCS 映射文件作为第一个 bq 表（外部表），并将您的BQ_Table作为另一个 bq 表。只需编写一个更新查询来更新每个映射文件Values中的字段。BQ_Table
  • 在 GCS 中的映射数据更新后，创建一个云函数以在预定查询之上运行。

推荐阅读