Как создать таблицы Bigtable и семейства столбцов в задании потока данных, если они не существуют

Question

У меня есть Задание облачного потока данных , которое записывает все в одну таблицу и одно семейство столбцов.Как изменить это задание для записи в несколько таблиц и семейств столбцов, которые могут существовать или не существовать?Например, если семейство таблиц или столбцов не существует, создайте его и запишите в него.

Answer 1

При использовании BigQueryIO вы можете сделать это:

 .apply("BQ-write", BigQueryIO.writeTableRows()
                    .to(tableSpec)
                    .withJsonSchema(schema)
                    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

Это волшебство: BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED

При использовании вы должны указать схему JSON таблицы, например:

{
  "fields": [
    {
      "mode": "NULLABLE",
      "name": "f1",
      "type": "STRING"
    },
    {
      "mode": "NULLABLE",
      "name": "error",
      "type": "f2"
    },
    {
      "mode": "NULLABLE",
      "name": "f3",
      "type": "STRING"
    }
  ]
}

и TableSpec

   TableReference tableSpec =
                new TableReference()
                        .setProjectId(projectId)
                        .setDatasetId(dataset)
                        .setTableId(errorTable);

Answer 2

Этот пример должен показать вам, как проверить, существует ли таблица, и создать ее, если нет.

Интерфейс Admin из этого примера кода также можно использовать дляполучить семейства столбцов для таблицы (getTableDescriptor) и при необходимости создать одно (addColumn).