Bigtable-BigQuery Импорт через DataFlow: 2 вопроса по разбиению таблиц и меток времени

Question

У меня есть работа в Dataflow импортировании данных из Bigtable в Bigquery с использованием встроенных API Dataflow для обоих. У меня есть два вопроса:

Вопрос 1: Если исходные данные в один большой стол в Bigtable, как я могу разделить его на множество суб- или небольших таблиц в BigQuery динамически на основе, скажем, учитывая Bigtable row-key известен только во время выполнения?

код Java в DataFlow выглядит следующим образом:

p.apply(Read.from(CloudBigtableIO.read(config))) 
     .apply(ParDo.of(new SomeDoFNonBTSourceData())) 
     .apply(BigQueryIO.Write 
        .to(PROJ_ID + ":" + BQ_DataSet + "." + BQ_TableName) 
        .withSchema(schema) 
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE) 
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)); 
     p.run(); 

Таким образом, поскольку BQ_TableName должен подаваться на уровне кода, как я могу предоставить это программно на основе того, что видно внутри SomeDoFNonBTSourceData, подобно диапазон значений текущего RowKey? Если RowKey является «ас», то TableA, если «DF», то TableB и т.д.

Вопрос 2: Что такое правильный способ экспортировать Bigtable Timestamp в BigQuery так, чтобы в конечном итоге восстановить его в удобной для восприятия человеком формате в BigQuery?

Функция processElement в DoFn выглядит следующим образом:

public void processElement(ProcessContext c) 
{ 
    String valA = new String(c.element().getColumnLatestCell(COL_FAM, COL_NAME).getValueArray()); 
    Long timeStamp = c.element().getColumnLatestCell(COL_FAM, COL_NAME).getTimestamp(); 

    tr.put("ColA", valA); 
    tr.put("TimeStamp",timeStamp); 
    c.output(tr); 
} 

И во время строительства трубопровода, установка BQ схемы для столбца TIMESTAMP выглядит следующим образом:

List<TableFieldSchema> fields = new ArrayList<>(); 
    fields.add(new TableFieldSchema().setName("ColA").setType("STRING")); 
    fields.add(new TableFieldSchema().setName("TimeStamp").setType("TIMESTAMP")); 
    schema = new TableSchema().setFields(fields); 

Так Bigtable метка времени похоже, имеет тип Long, и я пробовал как "TIMESTAMP", так и "INTEGER" типов для столбца TimeStamp назначения в BQ (похоже, что в BQ как таковой нет). В конечном счете, мне нужно использовать столбец TimeStamp в BQ как для предложений «по порядку», так и для отображения информации в удобочитаемой форме (дата и время). Похоже, что часть «порядок» работает нормально, но мне не удалось выполнить конечный результат в любом значении - либо получить ошибки при запуске, либо что-то еще нечитаемое.

Answer 1

Кстати, я ищу ответ на вопрос, подобный Вопросу 1 :).

Для второго вопроса, я думаю, вам сначала нужно подтвердить, что длинная метка времени действительно является отметкой времени UNIX, я всегда предполагал, что BQ может воспринимать это как временную метку без какого-либо преобразования.

Но вы можете попробовать это ...

Long longTimeStamp = 1408452095L; 

Date timeStamp = new Date(); 
timeStamp.setTime(longTimeStamp * 1000); 

tr.put("TimeStamp", timeStamp.toInstant().toString());