首页 文章

是否可以根据窗口元素的时间戳动态生成BigQuery表名？

提问于
浏览 次
5

例如，如果我有一个5分钟窗口的数据流流媒体作业从PubSub读取，我理解如果我将一个超过两天的时间戳分配给一个元素，将会有一个带有此元素的窗口，如果我使用的示例将每日表输出到BigQueryIO.java中描述的BigQuery，该作业将使用实际日期在BigQuery表中写入过去两天的元素 .

我想将过去的元素写入BigQuery表中，并使用窗口元素的时间戳而不是当前窗口的时间，是否可能？

现在我按照DataflowJavaSDK / sdk / src / main / java / com / google / cloud / dataflow / sdk / io / BigQueryIO.java中描述的示例：

PCollection<TableRow> quotes = ...
    quotes.apply(Window.<TableRow>info(CalendarWindows.days(1)))
       .apply(BigQueryIO.Write
         .named("Write")
         .withSchema(schema)
         .to(new SerializableFunction<BoundedWindow, String>() {
               public String apply(BoundedWindow window) {
                 String dayString = DateTimeFormat.forPattern("yyyy_MM_dd").parseDateTime(
                   ((DaysWindow) window).getStartDate());
                 return "my-project:output.output_table_" + dayString;
               }
             }));
java google-bigquery google-cloud-dataflow

2 回答

  • 5

    如果我理解正确，您希望确保根据元素（引号）的固有时间戳创建BigQuery表，而不是管道运行时的挂钟时间 .

    TL;DR the code should already do what you want ;如果不是，请发布更多详情 .

    更长的解释：Dataflow中处理的关键创新之一是 event-time processing . 这意味着Dataflow中的数据处理几乎完全与处理发生时分离 - 重要的是处理事件发生的时间 . 这是在批处理或流数据源上运行完全相同的代码的关键要素（例如，使用处理历史点击日志的相同代码处理实时用户点击事件） . 它还可以灵活处理迟到的数据 .

    有关Dataflow处理模型这方面的描述，请参阅The world beyond batch，"Event time vs. processing time"部分（整篇文章非常值得一读） . 有关更深入的说明，请参阅VLDB paper . 在windowing和triggers的官方文档中，这也以更面向用户的方式进行了描述 .

    因此， there is no such thing as a "current window" 因为管道可能同时处理在不同时间发生并属于不同窗口的许多不同事件 . 实际上，正如VLDB文件指出的那样，Dataflow管道执行的一个重要部分是"group elements by window" .

    在您展示的管道中，我们将使用记录上的provided timestamps将您要写入BigQuery的记录分组到窗口中，并将每个窗口写入其自己的表，如有必要，为新遇到的窗口创建表 . 如果后期数据到达窗口（请参阅有关窗口的文档和触发器以讨论后期数据），我们将附加到已有的表中 .

    回复于
  • 1

    上述代码不再适用于我 . 在Google文档中有updated example，但DaysWindow被IntervalWindow取代，后者对我有用：

    PCollection<TableRow> quotes = ...
 quotes.apply(Window.<TableRow>into(CalendarWindows.days(1)))
   .apply(BigQueryIO.Write
     .named("Write")
     .withSchema(schema)
     .to(new SerializableFunction<BoundedWindow, String>() {
       public String apply(BoundedWindow window) {
         // The cast below is safe because CalendarWindows.days(1) produces IntervalWindows.
         String dayString = DateTimeFormat.forPattern("yyyy_MM_dd")
              .withZone(DateTimeZone.UTC)
              .print(((IntervalWindow) window).start());
         return "my-project:output.output_table_" + dayString;
       }
     }));
    回复于

相关问题