Переименовать колонку при чтении записи с помощью Beam?

Question

• Я пытаюсь использовать существующий пример кода Google (PubSubToBigQuery.java), чтобы проанализировать StackDriver сообщения журнала и вставить их в BigQuery.

• Проблема заключается водно из имен полей журнала SD - "@type", которое не допускается BigQuery.Поэтому я создал эту таблицу в BigQuery с другим именем поля (mytest).

• Теперь, когда я запускаю PubSubToBigQuery.java, очевидно, я получаю сообщение об ошибке,

  Поле "@ type" не найдено.

• Как я могу переименовать имя столбца с "@type" на "mytype" внутри моего кода луча?

Answer 1

Если вы хотите просто сохранить журналы Stackdriver в BigQuery без изменений, вы можете использовать встроенную функцию экспорта Stackdriver и создать приемник для BigQuery: https://cloud.google.com/logging/docs/export/

Если каким-либо образом экспорт для вас невозможен,Вы можете изменить логику преобразования в Beam.

В этом случае PubSubToBigQuery.java BigQueryIO использует коллекцию TableRow PC в качестве входных данных для записи сообщений в BigQuery.PubsubMessageToTableRow PTransform выполняет преобразование из PubsubMessage в TableRow с некоторой обработкой ошибок.Вы можете добавить ParDo с пользовательским DoFn, который изменяет имя столбца в созданном TableRow.Метод элемента процесса может выглядеть примерно так:

 @ProcessElement
 public void processElement(@Element TableRow row, OutputReceiver<TableRow> outputReceiver) {

    TableRow clone = row.clone();
    Object value = clone.get("@type");
    clone.remove("@type");
    clone.set("mytype", value);
    outputReceiver.output(clone);
}

Если вы используете неизмененный PubSubToBigQuery.java, который я связал, вы можете применить этот ParDo на PCollection jsonToTableRowOut.get(TRANSFORM_OUT) где-то около строки 323 в коде.