Flink常用API之HDFS文件Source

Flink常用API之HDFS文件Source

package sourceimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironment/** * @Author yqq * @Date 2021/12/25 13:17 * @Version 1.0 */object HDFSFileSource { def main(args: Array[String]): Unit = { val ev = StreamExecutionEnvironment.getExecutionEnvironment ev.setParallelism(1) import org.apache.flink.streaming.api.scala._ //读取HDFS上读取文件 val stream: DataStream[String] = ev.readTextFile("hdfs://mycluster/wc.txt") //单词计算 stream.flatMap(_.split(" ")) .map((_,1)) .keyBy(0) .sum(1) .print() ev.execute("wordcount") }}

HDFS数据图

[root@node1 ~]# hdfs dfs -cat /wc.txt21/12/25 14:52:10 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicablehello tom andy joy hello rose hello joy mark andy hello tom andy rose hello joy

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。