Apache Beam WordCount编程实战及源码解读-阿里云开发者社区

开发者社区> 王小雷> 正文

Apache Beam WordCount编程实战及源码解读

简介: 概述:Apache Beam WordCount编程实战及源码解读,并通过intellij IDEA和terminal两种方式调试运行WordCount程序,Apache Beam对大数据的批处理和流处理,提供一套先进的统一的编程模型,并可以运行大数据处理引擎上。完整项目Github源码 负责公司大数据处理相关架构,但是具有多样性,极大的增加了开发成本,急需统一编程处理
+关注继续查看

概述:Apache Beam WordCount编程实战及源码解读,并通过intellij IDEA和terminal两种方式调试运行WordCount程序,Apache Beam对大数据的批处理和流处理,提供一套先进的统一的编程模型,并可以运行大数据处理引擎上。完整项目Github源码

Apache Beam WordCount编程实战及源码解读

负责公司大数据处理相关架构,但是具有多样性,极大的增加了开发成本,急需统一编程处理,Apache Beam,一处编程,处处运行,故将折腾成果分享出来。

1.Apache Beam编程实战–前言,Apache Beam的特点与关键概念。

Apache Beam 于2017年1月10日成为Apache新的顶级项目。

1.1.Apache Beam 特点:

  • 统一:对于批处理和流媒体用例使用单个编程模型。
  • 方便:支持多个pipelines环境运行,包括:Apache Apex, Apache Flink, Apache Spark, 和 Google Cloud Dataflow。
  • 可扩展:编写和分享新的SDKs,IO连接器和transformation库
    部分翻译摘自官网:Apacher Beam 官网

1.2.Apache Beam关键概念:

1.2.1.Apache Beam SDKs

主要是开发API,为批处理和流处理提供统一的编程模型。目前(2017)支持JAVA语言,而Python正在紧张开发中。

1.2.2. Apache Beam Pipeline Runners(Beam的执行器/执行者们),支持Apache Apex,Apache Flink,Apache Spark,Google Cloud Dataflow多个大数据计算框架。可谓是一处Apache Beam编程,多计算框架运行。

1.2.3. 他们的对如下的支持情况详见

Apache Beam WordCount编程实战及源码解读

2.Apache Beam编程实战–Apache Beam源码解读

基于maven,intellij IDEA,pom.xm查看 完整项目Github源码 。直接通过IDEA的项目导入功能即可导入完整项目,等待MAVEN下载依赖包,然后按照如下解读步骤即可顺利运行。

2.1.源码解析-Apache Beam 数据流处理原理解析:

关键步骤:

  • 创建Pipeline
  • 将转换应用于Pipeline
  • 读取输入文件
  • 应用ParDo转换
  • 应用SDK提供的转换(例如:Count)
  • 写出输出
  • 运行Pipeline

Apache Beam WordCount编程实战及源码解读

2.2.源码解析,完整项目Github源码,附WordCount,pom.xml等

/**
 * MIT.
 * Author: wangxiaolei(王小雷).
 * Date:17-2-20.
 * Project:ApacheBeamWordCount.
 */


import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;


public class WordCount {

    /**
     *1.a.通过Dofn编程Pipeline使得代码很简洁。b.对输入的文本做单词划分,输出。
     */
    static class ExtractWordsFn extends DoFn<String, String> {
        private final Aggregator<Long, Long> emptyLines =
                createAggregator("emptyLines", Sum.ofLongs());

        @ProcessElement
        public void processElement(ProcessContext c) {
            if (c.element().trim().isEmpty()) {
                emptyLines.addValue(1L);
            }

            // 将文本行划分为单词
            String[] words = c.element().split("[^a-zA-Z']+");
            // 输出PCollection中的单词
            for (String word : words) {
                if (!word.isEmpty()) {
                    c.output(word);
                }
            }
        }
    }

    /**
     *2.格式化输入的文本数据,将转换单词为并计数的打印字符串。
     */
    public static class FormatAsTextFn extends SimpleFunction<KV<String, Long>, String> {
        @Override
        public String apply(KV<String, Long> input) {
            return input.getKey() + ": " + input.getValue();
        }
    }
    /**
     *3.单词计数,PTransform(PCollection Transform)将PCollection的文本行转换成格式化的可计数单词。
     */
    public static class CountWords extends PTransform<PCollection<String>,
            PCollection<KV<String, Long>>> {
        @Override
        public PCollection<KV<String, Long>> expand(PCollection<String> lines) {

            // 将文本行转换成单个单词
            PCollection<String> words = lines.apply(
                    ParDo.of(new ExtractWordsFn()));

            // 计算每个单词次数
            PCollection<KV<String, Long>> wordCounts =
                    words.apply(Count.<String>perElement());

            return wordCounts;
        }
    }

    /**
     *4.可以自定义一些选项(Options),比如文件输入输出路径
     */
    public interface WordCountOptions extends PipelineOptions {

        /**
         * 文件输入选项,可以通过命令行传入路径参数,路径默认为gs://apache-beam-samples/shakespeare/kinglear.txt
         */
        @Description("Path of the file to read from")
        @Default.String("gs://apache-beam-samples/shakespeare/kinglear.txt")
        String getInputFile();
        void setInputFile(String value);

        /**
         * 设置结果文件输出路径,在intellij IDEA的运行设置选项中或者在命令行中指定输出文件路径,如./pom.xml
         */
        @Description("Path of the file to write to")
        @Required
        String getOutput();
        void setOutput(String value);
    }
    /**
     * 5.运行程序
     */
    public static void main(String[] args) {
        WordCountOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                .as(WordCountOptions.class);
        Pipeline p = Pipeline.create(options);

        p.apply("ReadLines", TextIO.Read.from(options.getInputFile()))
                .apply(new CountWords())
                .apply(MapElements.via(new FormatAsTextFn()))
                .apply("WriteCounts", TextIO.Write.to(options.getOutput()));

        p.run().waitUntilFinish();
    }
}

3.支持Spark,Flink,Apex等大数据数据框架来运行该WordCount程序。完整项目Github源码(推荐,注意pom.xml模块加载是否成功,在工具中开发大数据程序,利于调试,开发体验较好)

3.1.intellij IDEA(社区版)中Spark大数据框架运行Pipeline计算程序

  • Spark运行

    • 设置VM options

      -DPapex-runner
    • 设置Programe arguments

      --inputFile=pom.xml --output=counts

Apache Beam WordCount编程实战及源码解读

3.2.intellij IDEA(社区版)中Apex,Flink等支持的大数据框架均可运行WordCount的Pipeline计算程序,完整项目Github源码

  • Apex运行

    • 设置VM options

      -DPapex-runner
    • 设置Programe arguments

      --inputFile=pom.xml --output=counts
  • Flink运行等等

    • 设置VM options

      -DPflink-runner
    • 设置Programe arguments

      --inputFile=pom.xml --output=counts

4.终端运行(Terminal)(不推荐,第一次下载过程很慢,开发体验较差)

4.1.以下命令是下载官方示例源码,第一次运行下载较慢,如果失败了就多运行几次,(推荐下载,完整项目Github源码)直接用上述解读在intellij IDEA中运行。

mvn archetype:generate       -DarchetypeRepository=https://repository.apache.org/content/groups/snapshots       -DarchetypeGroupId=org.apache.beam       -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples       -DarchetypeVersion=LATEST       -DgroupId=org.example       -DartifactId=word-count-beam       -Dversion="0.1"       -Dpackage=org.apache.beam.examples       -DinteractiveMode=false

Apache Beam WordCount编程实战及源码解读

4.2.打包并运行

mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount      -Dexec.args="--runner=SparkRunner --inputFile=pom.xml --output=counts" -Pspark-runner

Apache Beam WordCount编程实战及源码解读

4.3.成功运行结果

4.3.1.显示运行成功

Apache Beam WordCount编程实战及源码解读

4.3.2.WordCount输出计算结果

这里写图片描述

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
史上最快! 10小时大数据入门实战(九)- 前沿技术拓展Spark,Flink,Beam
spark Spark 开发语言及运行模式介绍 Scala安装 下载 Scala ...
1430 0
1月28日云栖精选夜读 | 终于等到你!阿里正式向 Apache Flink 贡献 Blink 源码
如同我们去年12月在 Flink Forward China 峰会所约,阿里巴巴内部 Flink 版本 Blink 将于 2019 年 1 月底正式开源。今天,我们终于等到了这一刻。
3912 0
带你读《Photoshop+Dreamweaver淘宝天猫网店美工与广告设计一本通 : 实战版》之一:写在处理之前
本书是一本从网店装修实际需求出发编写的实例型教程,以Photoshop和Dreamweaver为软件环境,详细讲解了它们在网店装修中的应用,帮助读者快速掌握网店图像处理和页面排版的核心技能。本书内容实用,图文并茂,可操作性强,适合网店店长、网店美工等希望快速掌握网店装修和推广技术的读者阅读和学习,也可作为大专院校相关专业或电商培训机构的教学参考书。
314 0
Java 网络编程实战(一) - 简介
Java 网络编程实战(一) - 简介
13 0
《React Native移动开发实战》一一2.2 Git版本控制工具
本节书摘来自华章出版社《React Native移动开发实战》一 书中的第2章,第2.2节,作者:袁林 著 ,更多章节内容可以访问云栖社区“华章计算机”公众号查看。
1753 0
k8s与日志--journalbeat源码解读
前言 对于日志系统的重要性不言而喻,参照沪江的一篇关于日志系统的介绍,基本上日志数据在以下几方面具有非常重要的作用: 数据查找:通过检索日志信息,定位相应的 bug ,找出解决方案 服务诊断:通过对日志信息进行统计、分析,了解服务器的负荷和服务运行状态 数据分析:可以做进一步的数据分析,比如根据请求中的课程 id ,找出 TOP10 用户感兴趣课程 日志+大数据+AI的确有很多想象空间。
1225 0
《React Native移动开发实战》一一1.5 小试牛刀——更改React Native项目源码
本节书摘来自华章出版社《React Native移动开发实战》一 书中的第1章,第1.2节,作者:袁林 著 ,更多章节内容可以访问云栖社区“华章计算机”公众号查看。
1091 0
《React Native移动开发实战》一一2.1 开发具备的基础知识说明
本节书摘来自华章出版社《React Native移动开发实战》一 书中的第2章,第2.1节,作者:袁林 著 ,更多章节内容可以访问云栖社区“华章计算机”公众号查看。
946 0
+关注
王小雷
专注大数据,人工智能的多面手,对新兴的技术与知识充满了好奇与渴望!
128
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《Nacos架构&原理》
立即下载
《看见新力量:二》电子书
立即下载
云上自动化运维(CloudOps)白皮书
立即下载