在Flink[Scala]中使用RollingLink将使用Avro序列化的对象写入HDFS

Question

我正在尝试使用Flink中的RollingLink将序列化到AVRO到HDFS的案例类编写出来。为了使avro文件可由HDFS反序列化，我使用了包装FSDataOutputStream的DataFileWriter。当我试图在DataFileWriter和FSDataOutputStream之间同步以关闭HDFS上的数据文件时，会引发异常，并且我实际上会在其他每个文件中获取数据。在Flink writer实现中，有没有一种方法可以将fs流与Avro writer同步？

我尝试过使用DataFileWriter close（）flush（）sync（）fsync（），但都失败了。同步方法似乎表现最好。我也尝试过同步写入方法，这似乎有效，但仍然产生了一个错误，我无法验证是否所有数据都保存到文件中。

class AvroWriter[OutputContainer <: org.apache.avro.specific.SpecificRecordBase] extends Writer[OutputContainer] {
  val serialVersionUID = 1L
  var outputStream: FSDataOutputStream = null
  var outputWriter: DataFileWriter[OutputContainer] = null
  override def open(outStream: FSDataOutputStream): Unit = {
    if (outputStream != null) {
      throw new IllegalStateException("AvroWriter has already been opened.")
    }
    outputStream = outStream
    if(outputWriter == null) {
      val writer: DatumWriter[OutputContainer] = new SpecificDatumWriter[OutputContainer](OutputContainer.SCHEMA$)
      outputWriter = new DataFileWriter[OutputContainer](writer)
      outputWriter.create(OutputContainer.SCHEMA$, outStream)
    }
  }
  override def flush(): Unit = {}
  override def close(): Unit = {
    if(outputWriter != null) {
      outputWriter.sync()
    }
    outputStream = null
  }
  override def write(element: OutputContainer) = {
    if (outputStream == null) {
      throw new IllegalStateException("AvroWriter has not been opened.")
    }
    outputWriter.append(element)
  }
  override def duplicate(): AvroWriter[OutputContainer] = {
    new AvroWriter[OutputContainer]
  }
}

尝试用上面的代码运行RollingLink会出现以下异常：

java.lang.Exception: Could not forward element to next operator
        at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher.run(LegacyFetcher.java:222)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.run(FlinkKafkaConsumer08.java:316)
        at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:78)
        at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:56)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:224)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
        at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.RuntimeException: Could not forward element to next operator
        at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:354)
        at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:337)
        at org.apache.flink.streaming.api.operators.StreamSource$NonTimestampContext.collect(StreamSource.java:158)
        at org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.run(LegacyFetcher.java:664)
Caused by: java.nio.channels.ClosedChannelException
        at org.apache.hadoop.hdfs.DFSOutputStream.checkClosed(DFSOutputStream.java:1353)
        at org.apache.hadoop.fs.FSOutputSummer.write(FSOutputSummer.java:98)
        at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.write(FSDataOutputStream.java:58)
        at java.io.DataOutputStream.write(DataOutputStream.java:107)
        at org.apache.avro.file.DataFileWriter$BufferedFileOutputStream$PositionFilter.write(DataFileWriter.java:446)
        at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
        at java.io.BufferedOutputStream.write(BufferedOutputStream.java:121)
        at org.apache.avro.io.BufferedBinaryEncoder$OutputStreamSink.innerWrite(BufferedBinaryEncoder.java:216)
        at org.apache.avro.io.BufferedBinaryEncoder.writeFixed(BufferedBinaryEncoder.java:150)
        at org.apache.avro.file.DataFileStream$DataBlock.writeBlockTo(DataFileStream.java:366)
        at org.apache.avro.file.DataFileWriter.writeBlock(DataFileWriter.java:383)
        at org.apache.avro.file.DataFileWriter.sync(DataFileWriter.java:401)
        at pl.neptis.FlinkKafkaConsumer.utils.AvroWriter.close(AvroWriter.scala:36)
        at org.apache.flink.streaming.connectors.fs.RollingSink.closeCurrentPartFile(RollingSink.java:476)
        at org.apache.flink.streaming.connectors.fs.RollingSink.openNewPartFile(RollingSink.java:419)
        at org.apache.flink.streaming.connectors.fs.RollingSink.invoke(RollingSink.java:373)
        at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:39)
        at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:351)
        ... 3 more

Answer 1

我终于找到了解决方案。由于流由RollingLink管理，因此无法在实现Writer的类中关闭它。另一方面，如果DataFileWriter包装了一个流，并且应该将一个文件转储到hdfs，则需要进行一些同步或关闭。诀窍不是关闭DataFileWriter，而是同步它，然后通过给它赋值null来丢弃它（当你考虑Scala和函数式编程时，这不是一种很常用的方式，但嘿，Flink是用Java开发的）。所以这个简单的技巧解决了我的问题：

override def close(): Unit = {
    if(outputWriter != null) {
      outputWriter.sync()
    }
    outputWriter = null
    outputStream = null
  }