0

我正在使用 flink AvroOutputFormat 写一些数据，

val source: DataSet[Row] = environment.createInput(inputBuilder.finish)
val tableEnv: BatchTableEnvironment = new BatchTableEnvironment(environment, TableConfig.DEFAULT)
val table: Table = source.toTable(tableEnv)
val avroOutputFormat = new AvroOutputFormat[Row](classOf[Row])
avroOutputFormat.setCodec(AvroOutputFormat.Codec.NULL)
source.write(avroOutputFormat, "/Users/x/Documents/test_1.avro").setParallelism(1)
environment.execute()

这会将数据写入一个名为test_1.avro. 当我试图读取文件时，

val users = new AvroInputFormat[Row](new Path("/Users/x/Documents/test_1.avro"), classOf[Row])
val usersDS = environment.createInput(users)
usersDS.print()

这会将行打印为，

java.lang.Object@4462efe1,java.lang.Object@7c3e4b1a,java.lang.Object@2db4ad1,java.lang.Object@765d55d5,java.lang.Object@2513a118,java.lang.Object@2bfb583b,java.lang.Object@73ae0257,java.lang.Object@6fc1020a,java.lang.Object@5762658b

是否有可能打印此数据值而不是对象地址的方法。

4

1 回答 1

1

您正在以一种奇怪的方式混合 Table API 和 Datastream API。最好坚持使用一种 API 或使用适当的转换方法。

因为你基本上不会让 Flink 知道预期的输入/输出模式。classOf[Row]是一切，什么都不是。

要将表格写入 Avro 文件，请使用表格连接器。基本草图

tableEnv.connect(new FileSystem("/path/to/file"))
    .withFormat(new Avro().avroSchema("...")) // <- Adjust
    .withSchema(schema)
    .createTemporaryTable("AvroSinkTable")
table.insertInto("AvroSinkTable")

编辑：截至目前，不幸的是，文件系统连接器不支持 Avro。

所以别无选择，只能使用数据集 API。我建议使用avrohugger为您的 avro 模式生成适当的 scala 类。

// convert to your scala class
val dsTuple: DataSet[User] = tableEnv.toDataSet[User](table)
// write out
val avroOutputFormat = new AvroOutputFormat<>(User.class)
avroOutputFormat.setCodec(Codec.SNAPPY)
avroOutputFormat.setSchema(User.SCHEMA$)
specificUser.write(avroOutputFormat, outputPath1)
于 2020-02-24T09:07:21.610 回答