Flink, как создать таблицу со схемой, выведенной из входных данных Avro - PullRequest
Новая
       22

Flink, как создать таблицу со схемой, выведенной из входных данных Avro

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

Я загрузил файл Avro в наборе данных Flink: 

AvroInputFormat<GenericRecord> test = new AvroInputFormat<GenericRecord>(
        new Path("PathToAvroFile")
        , GenericRecord.class);
DataSet<GenericRecord> DS = env.createInput(test);

usersDS.print();

и вот результаты печати DS: 

{"N_NATIONKEY": 14, "N_NAME": "KENYA", "N_REGIONKEY": 0, "N_COMMENT": " pending excuses haggle furiously deposits. pending, express pinto beans wake fluffily past t"}
{"N_NATIONKEY": 15, "N_NAME": "MOROCCO", "N_REGIONKEY": 0, "N_COMMENT": "rns. blithely bold courts among the closely regular packages use furiously bold platelets?"}
{"N_NATIONKEY": 16, "N_NAME": "MOZAMBIQUE", "N_REGIONKEY": 0, "N_COMMENT": "s. ironic, unusual asymptotes wake blithely r"}
{"N_NATIONKEY": 17, "N_NAME": "PERU", "N_REGIONKEY": 1, "N_COMMENT": "platelets. blithely pending dependencies use fluffily across the even pinto beans. carefully silent accoun"}
{"N_NATIONKEY": 18, "N_NAME": "CHINA", "N_REGIONKEY": 2, "N_COMMENT": "c dependencies. furiously express notornis sleep slyly regular accounts. ideas sleep. depos"}
{"N_NATIONKEY": 19, "N_NAME": "ROMANIA", "N_REGIONKEY": 3, "N_COMMENT": "ular asymptotes are about the furious multipliers. express dependencies nag above the ironically ironic account"}
{"N_NATIONKEY": 20, "N_NAME": "SAUDI ARABIA", "N_REGIONKEY": 4, "N_COMMENT": "ts. silent requests haggle. closely express packages sleep across the blithely"}

Теперь я хочу создать таблицу изНабор данных DS с точно такой же схемой файла Avro, я имею в виду столбцы должны быть N_NATIONKEY, N_NAME, N_REGIONKEY и N_COMMENT.

Я знаю, используя строку: 

tableEnv.registerDataSet("tbTest", DS, "field1, field2, ...");

Я могу создатьтаблицу и установить столбцы, но я хочу, чтобы столбцы автоматически выводились из данных.Является ли это возможным?Кроме того, я попытался 

tableEnv.registerDataSet("tbTest", DS);

, но он создает таблицу со схемой: 

root
 |-- f0: GenericType<org.apache.avro.generic.GenericRecord>

1 Ответ

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

GenericRecord - это черный ящик для среды выполнения API таблиц и SQL, поскольку число полей и их тип данных не определены.Я бы порекомендовал использовать сгенерированный Avro класс, который расширяет SpecificRecord.Эти конкретные типы также распознаются системой типов Flink, и вы можете правильно обращаться к отдельным полям с правильным типом данных.

В качестве альтернативы, вы можете реализовать пользовательский UDF , который извлекает поля с правильным типом данных getAvroInt(f0, "myField"), getAvroString(f0, "myField") и т. Д.

Некоторые псевдокоды для этого: 

class AvroStringFieldExtract extends ScalarFunction {
    public String eval(GenericRecord r, String fieldName) {
        return r.get(fieldName).toString();
    }
}

tableEnv.registerFunction("getAvroFieldString", new AvroStringFieldExtract())
...