如何使用akka中的Source读取二进制avro fileData？

Question

我正在尝试使用来自 akka Streams 的 Source 读取 avro 文件。

akka 流中的源读取类似 FileIO.FromPath(File) 的数据，它将根据 (\n) 字符读取和分隔行，而 avro 是如何工作的？

流程：

    object AvroFlow  {
    def apply(jobDate: String): Flow[GenericRecord, GenericRecord, NotUsed] = {
            Flow[GenericRecord].map { rec => rec.put("date", "20190812") rec}       
    }
  }

图表：

object AvroRunner {
    def build (src: Source[GenericRecord, NotUsed],
                                     flw: Flow[GenericRecord, GenericRecord, NotUsed],
                                     snk:Flow[GenericRecord, Future[Done])
    : AvroRunner = {
      new AvroRunner(srtc,flw,snk)
    }
  }
class AvroRunner private(src: Source[GenericRecord, NotUsed],
                                     flw: Flow[GenericRecord, GenericRecord, NotUsed],
                                     snk:Flow[GenericRecord, Future[Done]){
  import scala.concurrent.ExecutionContext.Implicits.global
  val GraphRunner = RunnableGraph.fromGraph(GraphDSL.create() {implicit builder =>
    import GraphDSL.Implicits._
    src ~> flw ~> snk
    ClosedShape
  })
}

Answer 1

创建 avro 数据对象的 akka Source 的最简单方法不是来自原始二进制文件本身。相反，从 avro 库提供的DataFileReader 创建源。

从the documentation 我们首先从java.io.File 生成器创建文件阅读器：

def createFileReader[T : ClassTag](fileGenerator : () => File) : DataFileReader[T] = 
  new DataFileReader[T](file(), new SpecificDatumReader[T](classTag[T].runtimeClass))

这可以用来创建一个 scala Iterator:

def dataFileReaderToIterator[T](dataFileReader : DataFileReader[T]) : Iterator[T] = 
  new Iterator[T] {
    override def hasNext : Boolean = dataFileReader.hasNext

    override def next() : T = dataFileReader.next
  }

我们现在可以从文件生成器构造一个流 Source：

def fileToAvroSource[T](fileGenerator : () => File) : Source[T, _] = 
  Source.fromIterator[T](() => dataFileReaderToIterator[T](createFileReader(fileGenerator)))

背压？

似乎 avro 正在使用标准的 BufferedReader/OutputStream 技术来读取 File。因此，上述实现应该提供一直到文件源的背压。但是，我还没有确认是这种情况......