Как импортировать CSV файл в таблицу BigQuery без имен столбцов или схемы?

В настоящее время я пишу утилиту Java для импорта нескольких файлов CSV из GCS в BigQuery. Я легко могу достичь этого с помощью bq load, но я хотел сделать это с помощью задания Dataflow. Поэтому я использую Dataflow Pipeline и ParDo transformer (возвращает TableRow, чтобы применить его к BigQueryIO), и я создал StringToRowConverter() для преобразования. Здесь начинается актуальная проблема - мне нужно указать схему для таблицы назначения, хотя я не хочу создавать новую таблицу, если она не существует, - только пытается загрузить данные. Поэтому я не хочу вручную устанавливать имя столбца для TableRow, поскольку у меня около 600 столбцов.

public class StringToRowConverter extends DoFn<String, TableRow> {

private static Logger logger = LoggerFactory.getLogger(StringToRowConverter.class);

public void processElement(ProcessContext c) {
    TableRow row = new TableRow();          
    row.set("DO NOT KNOW THE COLUMN NAME", c.element());
    c.output(row);
}
}

Кроме того, предполагается, что таблица уже существует в наборе данных BigQuery, и мне не нужно ее создавать, а также файл CSV содержит столбцы в правильном порядке.

Если в этом сценарии нет обходного пути, а для загрузки данных требуется имя столбца, то я могу получить его в первой строке файла CSV.

Любая помощь будет оценена.

Ответы

Ответ 1

Чтобы избежать создания таблицы, вы должны использовать BigQueryIO.Write.CreateDisposition.CREATE_NEVER файла BigQueryIO.Write во время конфигурации конвейера. Источник: https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/BigQueryIO.Write

Вам не нужно заранее знать схему таблиц BigQuery, вы можете обнаружить ее динамически. Например, вы можете использовать API BigQuery (https://cloud.google.com/bigquery/docs/reference/rest/v2/tables/get), чтобы запросить схему таблицы и передать ее как параметр для класса StringToRowConverter. Другой вариант и предполагая, что первая строка является заголовком, состоит в том, чтобы пропустить первую строку и использовать ее для правильного отображения остальной части файла.

Нижеприведенный код реализует второй подход, а также настраивает вывод для добавления в существующую таблицу BigQuery.

public class DFJob {

    public static class StringToRowConverter extends DoFn<String, TableRow> {

        private String[] columnNames;

        private boolean isFirstRow = true;

        public void processElement(ProcessContext c) {
            TableRow row = new TableRow();

            String[] parts = c.element().split(",");

            if (isFirstRow) {
                columnNames = Arrays.copyOf(parts, parts.length);
                isFirstRow = false;
            } else {
                for (int i = 0; i < parts.length; i++) {
                    row.set(columnNames[i], parts[i]);
                }
                c.output(row);
            }
        }
    }

    public static void main(String[] args) {
        DataflowPipelineOptions options = PipelineOptionsFactory.create()
                .as(DataflowPipelineOptions.class);
        options.setRunner(BlockingDataflowPipelineRunner.class);

        Pipeline p = Pipeline.create(options);

        p.apply(TextIO.Read.from("gs://dataflow-samples/myfile.csv"))
                .apply(ParDo.of(new StringToRowConverter()))
                .apply(BigQueryIO.Write.to("myTable")
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

        PipelineResult result = p.run();
    }
}