0

我已经阅读了一些文档textFile，并做了flatMap一个单词，为每个单词添加了一些额外的信息：

val col = sc.textFile(args.getOrElse("input","documents/*"))
    .flatMap(_.split("\\s+").filter(_.nonEmpty))
val mapped = col.map(t => t + ": " + extraInformation())

我目前正在轻松地将其保存为文本

mapped.saveAsTextFile(args.getOrElse("output", "results"))

但我不知道如何将地图保存到 BigQuery 架构。我见过的所有示例都从 BigQuery 创建初始 Scollection，然后将其保存到另一个表中，因此初始集合[TableRow]不是[String].

这里的正确方法是什么？我是否应该调查如何将我的数据转换为 Big Query 可以接受的一种集合？或者我应该尝试进一步调查如何将这个纯文本直接推送到表格中？

4

2 回答 2

3

我建议@BigQueryType.toTable在案例类上使用注释，如下所示：

import com.spotify.scio.bigquery._

object MyScioJob {

  @BigQueryType.toTable
  case class WordAnnotated(word: String, extraInformation: String)


  def main(args: Array[String]): Unit = {
    // ...job setup logic

    sc.textFile(args.getOrElse("input","documents/*"))
      .flatMap(_.split("\\s+").filter(_.nonEmpty))
      .map(t => WordAnnotated(t, extraInformation())
      .saveAsTypedBigQuery("myProject:myDataset.myTable")
  }
}

Scio wiki上有更多关于此的信息。

于 2019-01-31T18:23:57.770 回答
0

为了写入 BigQuery，您需要定义一个 TableSchema：

public static final TableSchema BQ_TABLE_SCHEMA = new TableSchema();
public static final List<TableFieldSchema> BQ_FIELDS;

static {
    TableFieldSchema string_field = new TableFieldSchema()
            .setName("string_field")
            .setType(FieldType.STRING.toString())
            .setMode(FieldMode.NULLABLE.toString());

    BQ_FIELDS = Lists.newArrayList(
            string_field
    );

    BQ_TABLE_SCHEMA.setFields(BQ_FIELDS);
}

然后您需要将您的 String 转换为 TableRow 对象：

.apply("ConvertToTableRow", ParDo.of(new DoFn<String, TableRow>() {
                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        c.output(new TableRow().set("string_field", c.element()));
                    }
                }))
.apply("InsertTableRowsToBigQuery",
                        BigQueryIO.writeTableRows().to("project_id:dataset_name.table_name")
                                .withSchema(BQ_TABLE_SCHEMA)
                                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND))
                .getFailedInserts();

你也可以看一下Java中的这个例子，它与Scio中需要做的非常相似：https ://github.com/apache/beam/blob/master/examples/java/src/main/java /org/apache/beam/examples/complete/StreamingWordExtract.java#L78

于 2019-01-30T14:38:32.217 回答