title: Process continuously parquet files as Datastreams in Flink's DataStream API
body: <p>I have a parquet file on HDFS. It is overwritten daily with a new one. My goal is to emit this parquet file continuously - <strong>when it changes</strong> - as a DataStream in a Flink Job using the DataStream API. 
  The end goal is to use the file content in a Broadcast State, but this is out of scope for this question.</p>
  
  <ol>
  <li>To process a file <strong>continuously</strong>, there is this very useful API: <a href="https://ci.apache.org/projects/flink/flink-docs-stable/dev/datastream_api.html#data-sources" rel="nofollow noreferrer">Data-sources</a> about datasources. More specifically, <strong>FileProcessingMode.PROCESS_CONTINUOUSLY</strong>: this is exactly what I need. This works for reading/monitoring text files, no problem, but not for parquet files:</li>
  </ol>
  
  <pre><code>// Partial version 1: the raw file is processed continuously
  val path: String = "hdfs://hostname/path_to_file_dir/"
  val textInputFormat: TextInputFormat = new TextInputFormat(new Path(path))
  // monitor the file continuously every minute
  val stream: DataStream[String] = streamExecutionEnvironment.readFile(textInputFormat, path, FileProcessingMode.PROCESS_CONTINUOUSLY, 60000)
  </code></pre>
  
  <ol start="2">
  <li>To process <strong>parquet</strong> files, I can use <strong>Hadoop Input Formats</strong> using this API:
  <a href="https://ci.apache.org/projects/flink/flink-docs-stable/dev/batch/hadoop_compatibility.html#using-hadoop-inputformats" rel="nofollow noreferrer">using-hadoop-inputformats</a>. However there is no FileProcessingMode parameter via this API, and this processes the file only once:</li>
  </ol>
  
  <pre><code>// Partial version 2: the parquet file is only processed once
  val parquetPath: String = "/path_to_file_dir/parquet_0000"
  // raw text format
  val hadoopInputFormat: HadoopInputFormat[Void, ArrayWritable] = HadoopInputs.readHadoopFile(new MapredParquetInputFormat(), classOf[Void], classOf[ArrayWritable], parquetPath)
  val stream: DataStream[(Void, ArrayWritable)] = streamExecutionEnvironment.createInput(hadoopInputFormat).map { record =&gt;
    // process the record here ...
  }
  </code></pre>
  
  <p>I would like to somehow combine the two APIs, to process continuously Parquet Files via the DataStream API. Have any of you tried something like this ?</p>