class Main {
def startApp(): Unit ={
val env = ExecutionEnvironment.getExecutionEnvironment
val job = Job.getInstance()
val dIf = new HadoopInputFormat[Void, AvroTamAlert](new AvroParquetInputFormat(), classOf[Void], classOf[AvroTamAlert], job)
FileInputFormat.addInputPath(job, new Path("/user/hive/warehouse/testpath"))
val dataset = env.createInput(dIf)
println(dataset.count())
env.execute("start hdfs parquet test")
}
}
object Main {
def main(args:Array[String]):Unit = {
new Main().startApp()
}
}
1条答案
按热度按时间vnzz0bqm1#
好 啊。我已经找到了一种通过apacheflink读取hdfs中parquet文件的方法。
您应该在pom.xml中添加以下依赖项
创建一个avsc文件来定义模式。实验:
运行“java-jard:\avro-tools-1.8.2.jar compile schema alert.avsc.”生成java类并将avrotamalert.java复制到项目中。
使用avroparquetinputformat读取hdfs中的Parquet文件: