如何过滤 bigquery 的重复表行数据以删除重复行？

Posted 2023-03-31

【中文标题】如何过滤 bigquery 的重复表行数据以删除重复行？

【英文标题】：How can I filter duplicate TableRow data for big query to remove duplicate rows?

【发布时间】：2019-11-20 05:31:00

【问题描述】：

我是 Dataflow 的新手，如果我的问题很有趣，请原谅我，我正在阅读一个 csv 文件并且它有重复的行，我正在读取这些数据并写入大查询，但是我不想重复数据到我的 BQ 表。

我想到了一种方法，但我不知道如何实现它，它涉及向架构添加某种标志以标记它唯一但我不知道如何

Lists.newArrayList(
  new TableFieldSchema()
         .setName("person_id")
         .setMode("NULLABLE").setType("STRING"),
  new TableFieldSchema()
         .setName("person_name")
         .setMode("NULLABLE")
         .setType("STRING") // Cant I add another unique property here?
) 

不知道该方法是否可行，但我只需要过滤从转换中检索到的行，例如

PCollection<TableRow> peopleRows = 
  pipeline
     .apply(
        "Convert to BiqQuery Table Row",
        ParDo.of(new FormatForBigquery())

    // Next step to filter duplicates

【参考方案1】：

如果我们将用于读取 CSV 的输出视为 PCollection，那么我们可以通过 Distinct 转换传递 PCollection 来消除重复项。此转换的目的是获取输入 PCollection 并生成一个新的 PCollection，它是没有重复的原始 PCollection。作为 Distinct 预制转换的一部分，有机会指定一个自己的函数，该函数将被调用以确定将两个 PCollection 对象分类为相等的类别以及删除哪些对象。

【参考方案2】：

您可以直接在 bigquery 中执行此操作，加载整个文件并使用目标表作为源表运行如下查询。无需数据流。

WITH cte as (
  SELECT
    ROW_NUMBER() over (PARTITION BY column1,column2,column3,...) as idx,
    *
  FROM my_table
)
SELECT
*
FROM cte
WHERE idx = 1