Поместите SCollection из textFile в BigQuery с Scio - PullRequest
Новая
       9

Поместите SCollection из textFile в BigQuery с Scio

0 голосов
/ 29 января 2019

Я прочитал некоторые документы с 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.Все примеры, которые я видел, создают исходную коллекцию Scollection из BigQuery, а затем сохраняют ее в другой таблице, так что исходная коллекция будет [TableRow] вместо [String].

Каков здесь правильный подход?Должен ли я выяснить, как преобразовать мои данные в коллекцию, которую может принять Big Query?Или я должен попытаться выяснить, как вставить этот простой текст прямо в таблицу?

Ответы [ 2 ]

0 голосов
/ 31 января 2019

Я бы предложил использовать аннотацию @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 .

0 голосов
/ 30 января 2019

Чтобы написать в 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

...