使用flink获取数据流的文件名

zz2j4svz 于 2021-06-25 发布在 Flink

关注(0)|答案(1)|浏览(502)

我有一个流式处理与Flink与csv文件在一个单一的路径。我想知道每个处理过的文件的文件名。
我目前正在使用此函数将csv文件读入路径（datapath）。

val recs:DataStream[CallCenterEvent] = env
          .readFile[CallCenterEvent](
          CsvReader.getReaderFormat[CallCenterEvent](dataPath, c._2),
          dataPath,
          FileProcessingMode.PROCESS_CONTINUOUSLY,
          c._2.fileInterval)
          .uid("source-%s-%s".format(systemConfig.name, c._1))
          .name("%s records reading".format(c._1))

并使用此函数获取tuplecsvinputformat。

def getReaderFormat[T <: Product : ClassTag : TypeInformation](dataPath:String, conf:URMConfiguration): TupleCsvInputFormat[T] = {
  val typeInfo = implicitly[TypeInformation[T]]
  val format: TupleCsvInputFormat[T] = new TupleCsvInputFormat[T](new Path(dataPath), typeInfo.asInstanceOf[CaseClassTypeInfo[T]])
  if (conf.quoteCharacter != null && !conf.quoteCharacter.equals(""))
    format.enableQuotedStringParsing(conf.quoteCharacter.charAt(0))
  format.setFieldDelimiter(conf.fieldDelimiter)
  format.setSkipFirstLineAsHeader(conf.ignoreFirstLine)
  format.setLenient(true)

  return format
}

进程运行正常，但我找不到方法来获取每个csv文件的文件名。
提前谢谢

scala streaming apache-flink flink-streaming data-stream

来源：https://stackoverflow.com/questions/46671411/get-file-name-of-datastream-with-flink

1条答案

按热度按时间

mnemlml81#

我遇到了类似的情况，我需要知道正在处理的记录的文件名。文件名中有些信息在记录中不可用。要求客户更改记录模式不是一个选项。
我找到了一种方法来访问底层源代码。在我的例子中，它是fileinputsplit（它有源数据文件的路径信息）

class MyTextInputFormat(p:Path ) extends TextInputFormat(p) {

     override def readRecord(reusable: String, bytes: Array[Byte], offset: Int, numBytes: Int):String = {
val fileName = {
      if (this.currentSplit != null)      
        this.currentSplit.getPath.getName
      else
         "unknown-file-path"
    }

    //Add FileName to the record!
    super.readRecord(reusable, bytes, offset, numBytes)+","+fileName
  }
}

现在，您可以在流设置中使用它

val format = new MyTextInputFormat(new Path(srcDir))
format.setDelimiter(prfl.lineSep)
val stream = env.readFile(format, srcDir, FileProcessingMode.PROCESS_CONTINUOUSLY, Time.seconds(10).toMilliseconds

虽然我的情况略有不同，但这种方法也应该对你有所帮助！

赞(0）回复(0）举报 2021-06-25

我来回答

使用flink获取数据流的文件名

1条答案

相关问题

热门标签

最新问答