Spark组件之Spark Streaming学习1--NetworkWordCount学习
2016-04-26 17:08
411 查看
更多代码请见:https://github.com/xubo245/SparkLearning
NetworkWordCount:每个1秒将接收的数据进行wordCount,不累加
使用
1.方法1:在集群的examples中启动
一个terminal:
可以在这个terminal发送数据,前面一个terminal就会统计信息
另一个terminal:
2.运行方法2:打成jar包上传运行:
运行脚本:
然后一个ternimal运行nc,一个运行这个脚本,同上
输入数据:
结果输出:
代码:
参考:
【1】 http://spark.apache.org/docs/1.5.2/streaming-programming-guide.html
NetworkWordCount:每个1秒将接收的数据进行wordCount,不累加
使用
1.方法1:在集群的examples中启动
一个terminal:
nc -lk 9999
可以在这个terminal发送数据,前面一个terminal就会统计信息
另一个terminal:
./bin/run-example streaming.NetworkWordCount localhost 9999
2.运行方法2:打成jar包上传运行:
运行脚本:
#!/usr/bin/env bash spark-submit --name WordCountSpark \ --class org.apache.spark.Streaming.learning.NetworkWordCount \ --master spark://<strong>Master</strong>:7077 \ --executor-memory 512M \ --total-executor-cores 10 Streaming.jar localhost 9999
然后一个ternimal运行nc,一个运行这个脚本,同上
输入数据:
hadoop@Master:~$ sudo nc -lk 9999 a hello world a hello world hello hw^Hello word a a a a a a a
结果输出:
hadoop@Master:~/cloud/testByXubo/spark/Streaming$ ./submitJob.sh ------------------------------------------- Time: 1461661853000 ms ------------------------------------------- ------------------------------------------- Time: 1461661854000 ms ------------------------------------------- (,1) (hello,1) (world,1) (a,1) ------------------------------------------- Time: 1461661855000 ms ------------------------------------------- (a,1) ------------------------------------------- Time: 1461661856000 ms ------------------------------------------- ------------------------------------------- Time: 1461661857000 ms ------------------------------------------- (hello,1) ------------------------------------------- Time: 1461661858000 ms ------------------------------------------- (world,1) ------------------------------------------- Time: 1461661859000 ms ------------------------------------------- ------------------------------------------- Time: 1461661860000 ms ------------------------------------------- (hello,1) ------------------------------------------- Time: 1461661861000 ms ------------------------------------------- ------------------------------------------- Time: 1461661862000 ms ------------------------------------------- (hello,1) ------------------------------------------- Time: 1461661863000 ms ------------------------------------------- (word,1) ------------------------------------------- Time: 1461661864000 ms ------------------------------------------- (a,5) ------------------------------------------- Time: 1461661865000 ms -------------------------------------------
代码:
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // scalastyle:off println package org.apache.spark.Streaming.learning import org.apache.spark.SparkConf import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.Seconds import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions /** * Counts words in UTF8 encoded, '\n' delimited text received from the network every second. * * Usage: NetworkWordCount <hostname> <port> * <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive data. * * To run this on your local machine, you need to first run a Netcat server * `$ nc -lk 9999` * and then run the example * `$ bin/run-example org.apache.spark.examples.streaming.NetworkWordCount localhost 9999` */ object NetworkWordCount { def main(args: Array[String]) { if (args.length < 2) { System.err.println("Usage: NetworkWordCount <hostname> <port>") System.exit(1) } StreamingExamples.setStreamingLogLevels() // Create the context with a 1 second batch size val sparkConf = new SparkConf().setAppName("NetworkWordCount") val ssc = new StreamingContext(sparkConf, Seconds(1)) // Create a socket stream on target ip:port and count the // words in input stream of \n delimited text (eg. generated by 'nc') // Note that no duplication in storage level only for running locally. // Replication necessary in distributed scenario for fault tolerance. val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER) val words = lines.flatMap(_.split(" ")) val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _) wordCounts.print() ssc.start() ssc.awaitTermination() } } // scalastyle:on println
参考:
【1】 http://spark.apache.org/docs/1.5.2/streaming-programming-guide.html
相关文章推荐
- Java IO与NIO的一些文件拷贝测试
- Stream、WshShell、WshUrlShortcut对象及Shell.Application的参数与使用
- Node.js中的流(Stream)介绍
- php中stream(流)的用法
- php错误提示failed to open stream: HTTP request failed!的完美解决方法
- C# Stream 和 byte[] 之间的转换
- 浅析Node.js 中 Stream API 的使用
- Nodejs Stream 数据流使用手册
- 利用stream实现一个简单的http下载器
- 浅谈PHP中Stream(流)
- php常用Stream函数集介绍
- pgpool-II+Hot_standby+Streaming replication环境搭建
- java8新特性(2)-Stream
- Scala: 一次命令式到函数式的重构
- PHP使用stream_context_create()模拟POST/GET请求的方法
- Nodejs Stream 数据流使用手册
- 标签 stream_context_create - 第1页 -- 简明现代魔法
- Oracle 10g R2 Stream环境的监控管理总结
- Oracle 10g R2 Stream复制环境的归档管理总结