文档章节

基于HDFS的实时wordcount程序

别寒
 别寒
发布于 2017/08/02 14:41
字数 219
阅读 5
收藏 0
package cn.hhb.spark.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by dell on 2017/8/2.
 */
public class HDFSWordCount {

    public static void main(String[] args) {

        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("HDFSWordCount").setMaster("local[2]")
                .set("spark.testing.memory", "2147480000");

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

        JavaDStream<String> lines = jssc.textFileStream("hdfs://spark1:9000/wordcount_dir");

        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" "));
            }
        });

        JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        wordCounts.print();

        jssc.start();
        jssc.awaitTermination();
        jssc.close();
    }
}

scala版:

package com.hhb.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by dell on 2017/8/2.
  */
object HDFSWordCount {
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setAppName("HDFSWordCount")
      .setMaster("local[2]")
      .set("spark.testing.memory", "2147480000")
    val ssc = new StreamingContext(conf, Seconds(5))

    val lines = ssc.textFileStream("hdfs://spark1:9000/wordcount_dir")
    val words = lines.flatMap(_.split(" "))
    val pairs = words.map((_, 1))
    val wordCounts = pairs.reduceByKey(_ + _)

    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()

  }
}

© 著作权归作者所有

共有 人打赏支持
别寒
粉丝 30
博文 270
码字总数 136713
作品 0
永州
程序员
7.Spark Streaming:输入DStream之基础数据源以及基于HDFS的实时wordcount程序

输入DStream之基础数据源 HDFS文件 基于HDFS文件的实时计算,其实就是,监控一个HDFS目录,只要其中有新文件出现,就实时处理。相当于处理实时的文件流。 streamingContext.fileStream(data...

weixin_32265569
2017/11/16
0
0
SparkStreaming+Kafka 实现统计基于缓存的实时uv

我的原创地址:https://dongkelun.com/2018/06/25/KafkaUV/ 前言 本文利用SparkStreaming+Kafka实现实时的统计uv,即独立访客,一个用户一天内访问多次算一次,这个看起来要对用户去重,其实...

董可伦
07/12
0
0
SparkStreaming+Kafka 实现基于缓存的实时wordcount程序

我的原创地址:https://dongkelun.com/2018/06/14/updateStateBykeyWordCount/ 前言 本文利用SparkStreaming和Kafka实现基于缓存的实时wordcount程序,什么意思呢,因为一般的SparkStreamin...

董可伦
07/12
0
0
11.updateStateByKey以及基于缓存的实时wordcount程序

updateStateByKey updateStateByKey操作,可以让我们为每个key维护一份state,并持续不断的更新该state。 1、首先,要定义一个state,可以是任意的数据类型; 2、其次,要定义state更新函数—...

weixin_32265569
2017/11/18
0
0
写给大数据开发初学者的话 | 附教程

导读: 第一章:初识Hadoop 第二章:更高效的WordCount 第三章:把别处的数据搞到Hadoop上 第四章:把Hadoop上的数据搞到别处去 第五章:快一点吧,我的SQL 第六章:一夫多妻制 第七章:越来...

小数点
2017/12/07
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

OSChina 周日乱弹 —— 种族不同,禁止交往

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @小小编辑:推荐歌曲《苏菲小姐》- 鱼果 《苏菲小姐》- 鱼果 手机党少年们想听歌,请使劲儿戳(这里) @貓夏:下大雨 正是睡觉的好时候 临睡前...

小小编辑
今天
226
6
Python 搭建简单服务器

Python动态服务器网页(需要使用WSGI接口),基本实现步骤如下: 1.等待客户端的链接,服务器会收到一个http协议的请求数据报 2.利用正则表达式对这个请求数据报进行解析(请求方式、提取出文...

代码打碟手
今天
1
0
Confluence 6 删除垃圾内容

属性(profile)垃圾 属性垃圾的定义为,一个垃圾用户在 Confluence 创建了用户,但是这个用户在自己的属性页面中添加了垃圾 URL。 如果你有很多垃圾用户在你的系统中创建了属性,你可以使用...

honeymose
今天
1
0
qduoj~前端~二次开发~打包docker镜像并上传到阿里云容器镜像仓库

上一篇文章https://my.oschina.net/finchxu/blog/1930017记录了怎么在本地修改前端,现在我要把我的修改添加到部署到本地的前端的docker容器中,然后打包这个容器成为一个本地镜像,然后把这...

虚拟世界的懒猫
今天
1
0
UML中 的各种符号含义

Class Notation A class notation consists of three parts: Class Name The name of the class appears in the first partition. Class Attributes Attributes are shown in the second par......

hutaishi
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部