我相信你需要为
HCatRecord
这将是:
Pipeline p = Pipeline.create(options);
p.getCoderRegistry()
.registerCoderForClass(HCatRecord.class, WritableCoder.of(DefaultHCatRecord.class));
为了测试这个,我在
HCatalogIOTest
梁项目的类。它使用不同的模式,但应该演示一个完整的示例:
@Test
@NeedsEmptyTestTables
public void testSOKricket() {
// Register the coder
defaultPipeline
.getCoderRegistry()
.registerCoderForClass(HCatRecord.class, WritableCoder.of(DefaultHCatRecord.class));
defaultPipeline
.apply(TextIO.read().from("/tmp/words.txt"))
.apply(ParDo.of(new PukeHive()))
.apply(
HCatalogIO.write()
.withConfigProperties(getConfigPropertiesAsMap(service.getHiveConf()))
.withDatabase(TEST_DATABASE)
.withTable(TEST_TABLE)
.withPartition(new java.util.HashMap<>())
.withBatchSize(1L));
defaultPipeline.run();
}
static class PukeHive extends DoFn<String, HCatRecord> {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
// our test schema is (mycol1 string, mycol2 int)
DefaultHCatRecord rec = new DefaultHCatRecord(2);
rec.set(0, c.element());
rec.set(1, 1);
c.output(rec);
}
}
