Spark Streaming+IntelliJ Idea+Maven开发环境搭建 - 推酷
Spark Streaming+IntelliJ Idea+Maven开发环境搭建
本博客文章如无特别说明，均为原创！转载请注明出处：Big data enthusiast(
本文链接地址:
2 选择 Create from archivetype –& scala-archetype-simple
3 配置好GroupId和ArtifactId，下一步、下一步
4 pom.xml文件配置如下，需要引入依赖
5 log4J.properties配置信息如下
6 工程整体的目录结构如下所示
7 选择菜单栏run-edit configurations
9 选择菜单栏file project structure进行打包配置
国内关于Spark流处理方面的资料实在是少之又少，开发环境搭建上一些细节上的说明就更少了，本文主要介绍在Windows下通过IntelliJ Idea连接远程服务器的Spark节点，接收FlumeNG收集的日志数据实现实时的数据处理。开发语言为Scala。
这里我们假设已经部署好Spark 1.5.2集群，并且集群的运行模式为Standalone HA，假设已经存在FlumeNG的Agent发过来的实时数据流。
新建MAVEN工程
选择 Create from archivetype –& scala-archetype-simple
配置好GroupId和ArtifactId，下一步、下一步
pom.xml文件配置如下，需要引入依赖
flume-ng-sdk、
-streaming-flume_2.10、spark-streaming_2.10、jackson.core、jackson-databind、jackson-module-scala_2.10：
&project xmlns=&http://maven.apache.org/POM/4.0.0& xmlns:xsi=&http://www.w3.org/2001/XMLSchema-instance&
xsi:schemaLocation=&http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd&&
&modelVersion&4.0.0&/modelVersion&
&groupId&com.changtu&/groupId&
&artifactId&spark&/artifactId&
&version&1.0-SNAPSHOT&/version&
&inceptionYear&2008&/inceptionYear&
&properties&
&scala.version&2.10.6&/scala.version&
&/properties&
&repositories&
&repository&
&id&scala-tools.org&/id&
&name&Scala-Tools Maven2 Repository&/name&
&url&http://scala-tools.org/repo-releases&/url&
&/repository&
&/repositories&
&pluginRepositories&
&pluginRepository&
&id&scala-tools.org&/id&
&name&Scala-Tools Maven2 Repository&/name&
&url&http://scala-tools.org/repo-releases&/url&
&/pluginRepository&
&/pluginRepositories&
&dependencies&
&dependency&
&groupId&org.scala-lang&/groupId&
&artifactId&scala-library&/artifactId&
&version&${scala.version}&/version&
&/dependency&
&dependency&
&groupId&junit&/groupId&
&artifactId&junit&/artifactId&
&version&3.8.1&/version&
&scope&test&/scope&
&/dependency&
&dependency&
&groupId&org.apache.flume&/groupId&
&artifactId&flume-ng-sdk&/artifactId&
&version&1.5.2&/version&
&/dependency&
&dependency&
&groupId&org.apache.spark&/groupId&
&artifactId&spark-streaming-flume_2.10&/artifactId&
&version&1.5.2&/version&
&/dependency&
&dependency&
&groupId&org.apache.spark&/groupId&
&artifactId&spark-streaming_2.10&/artifactId&
&version&1.5.2&/version&
&/dependency&
&dependency&
&groupId&com.fasterxml.jackson.core&/groupId&
&artifactId&jackson-core&/artifactId&
&version&2.4.4&/version&
&/dependency&
&dependency&
&groupId&com.fasterxml.jackson.core&/groupId&
&artifactId&jackson-databind&/artifactId&
&version&2.4.4&/version&
&/dependency&
&dependency&
&groupId&com.fasterxml.jackson.module&/groupId&
&artifactId&jackson-module-scala_2.10&/artifactId&
&version&2.4.4&/version&
&/dependency&
&/dependencies&
&sourceDirectory&src/main/scala&/sourceDirectory&
&testSourceDirectory&src/test/scala&/testSourceDirectory&
&groupId&org.scala-tools&/groupId&
&artifactId&maven-scala-plugin&/artifactId&
&executions&
&execution&
&goal&compile&/goal&
&goal&testCompile&/goal&
&/execution&
&/executions&
&configuration&
&scalaVersion&${scala.version}&/scalaVersion&
&arg&-target:jvm-1.7&/arg&
&/configuration&
&groupId&org.apache.maven.plugins&/groupId&
&artifactId&maven-eclipse-plugin&/artifactId&
&configuration&
&downloadSources&true&/downloadSources&
&buildcommands&
&buildcommand&ch.epfl.lamp.sdt.core.scalabuilder&/buildcommand&
&/buildcommands&
&additionalProjectnatures&
&projectnature&ch.epfl.lamp.sdt.core.scalanature&/projectnature&
&/additionalProjectnatures&
&classpathContainers&
&classpathContainer&org.eclipse.jdt.launching.JRE_CONTAINER&/classpathContainer&
&classpathContainer&ch.epfl.lamp.sdt.launching.SCALA_CONTAINER&/classpathContainer&
&/classpathContainers&
&/configuration&
&/plugins&
&reporting&
&groupId&org.scala-tools&/groupId&
&artifactId&maven-scala-plugin&/artifactId&
&configuration&
&scalaVersion&${scala.version}&/scalaVersion&
&/configuration&
&/plugins&
&/reporting&
&/project&
log4J.properties配置信息如下
log4j.rootLogger=WARN,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n
新建Scala object
package com.changtu
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
object IPHandler {
* 使用淘宝的REST接口获取IP数据, 返回JSON数据
* @param ipAddr ip地址
def getIPJSON(ipAddr: String): String = {
Source.fromURL(&/service/getIpInfo.php?ip=& + ipAddr).mkString
* 使用淘宝的REST接口获取IP数据, 返回运营商信息
* @param ipAddr ip地址
def main(args: Array[String]) {
val conf = new SparkConf().setAppName(&FlumeNG sink&)
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(2))
val stream = FlumeUtils.createStream(ssc, &xxx.xxx.xxx(配置FlumeNG Agent的IP)&, 22221, StorageLevel.MEMORY_AND_DISK)
stream.map(e =& &FlumeNG:header:& + e.event.get(0).toString + &body: & + new String(e.event.getBody.array)).print()
ssc.start()
ssc.awaitTermination()
工程整体的目录结构如下所示
选择菜单栏run-edit configurations
添加配置信息
Main class
org.apache.spark.deploy.SparkSubmit
Program arguments
com.changtu.IPHandler
E:\IdeaProjects\jars\spark-streaming-flume_2.10-1.5.2.jar,E:\IdeaProjects\jars\flume-ng-sdk-1.5.2.jar
spark://tts.node4:7077
E:\IdeaProjects\spark\out\artifacts\changtu\changtu.jar
选择菜单栏file project structure进行打包配置
-------------------------------------------
Time: 0 ms
-------------------------------------------
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=8}body:
11:15:36 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGES0
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=5}body:
11:15:37 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE1
-------------------------------------------
Time: 0 ms
-------------------------------------------
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=7}body:
11:15:38 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE2
&span style=&font-size: 10 font-family: 'courier new', courier,&&FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=4}body:
11:15:39 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE3&/span&
-------------------------------------------
Time: 0 ms
-------------------------------------------
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=6}body:
11:15:40 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE4
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=8}body:
11:15:41 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE5
-------------------------------------------
Time: 0 ms
-------------------------------------------
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=0}body:
11:15:42 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE6
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=2}body:
11:15:43 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE7
-------------------------------------------
Time: 0 ms
-------------------------------------------
FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=4}body:
11:15:44 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE8
&span style=&font-size: 10 font-family: 'courier new', courier,&&FlumeNG:header:{flume.client.log4j.log.level=20000, flume.client.log4j.logger.name=com.changtu.datapush.DBToolkit, flume.client.log4j.message.encoding=UTF8, flume.client.log4j.timestamp=6}body:
11:15:45 INFO [com.changtu.datapush.DBToolkit:22] - FlumeNG MESSAGE9&/span&
如果要打包放到服务器调用，可以通过以下命令调用
spark-submit --name &spark-flume& --master spark://tts.node4:7077 --class com.changtu.IPHandler --jars /appl/scripts/spark-streaming-flume_2.10-1.5.2.jar,/appl/scripts/flume-ng-sdk-1.5.2.jar --executor-memory 300m /appl/scripts/changtu.jar
已发表评论数()
请填写推刊名
描述不能大于100个字符!
权限设置： 公开
仅自己可见
正文不准确
标题不准确
排版有问题
主题不准确
没有分页内容
图片无法显示
视频无法显示
与原文不一致基于Intellij&IDEA的Spark开发-Scala(2)
Spark Application目前部署的主要方式是spark-submit。其中，
普通命令行方式：
bin/spark-submit&--class&spark.FlumeEventPrint&--jars&/Users/caocao/bi_architecure_research/spark/spark-streaming-flume_2.10-1.2.0.jar,/Users/caocao/workstation/workspace_idea/apache-flume-1.5.2-bin/lib/flume-avro-source-1.5.2.jar,/Users/caocao/workstation/workspace_idea/apache-flume-1.5.2-bin/lib/flume-ng-sdk-1.5.2.jar
spark://localhost:7077&/Users/caocao/bi_architecure_research/spark/projects/idea_test/out/artifacts/idea_test_jar/idea_test.jar
IDE开发模式（重点）：
图A是新建的run application，其中：
class:&org.apache.spark.deploy.SparkSubmit
Program arguments：配置重点(图B)。
--class spark.FlumeEventPrint --jars
/Users/caocao/bi_architecure_research/spark/spark-streaming-flume_2.10-1.2.0.jar,/Users/caocao/workstation/workspace_idea/apache-flume-1.5.2-bin/lib/flume-avro-source-1.5.2.jar,/Users/caocao/workstation/workspace_idea/apache-flume-1.5.2-bin/lib/flume-ng-sdk-1.5.2.jar
--master spark://192.168.11.56:7077
/Users/caocao/bi_architecure_research/spark/projects/idea_test/out/artifacts/idea_test_jar/idea_test.jar
Before launch: &
Build 'idea_test.jar' artifact (图C)
基本思路是：每次edit 保存后，make，打包，然后run spark-submit部署到spark上。
参考：https://spark.apache.org/docs/1.2.0/configuration.html
已投稿到：
以上网友发言只代表其个人观点，不代表新浪网的观点或立场。10006人阅读
IDEA中如何给main方法附带参数&
public class Test{
public static void main(String args[]) {
& & & & if (args.length == 0) {
&& & & & &
& & & & & & throw new IllegalArgumentException(&Give a file name&);
& & & & new Test().run(args[0]);
& & } catch (IllegalArgumentException e) {
& & & & System.err.println(e.getMessage());
在命令后者终端工具中：
先编译javac Test.java
再运行java Test arg1 arg2 arg3 ……（参数之间用空格隔开）
在IDEA中：
点击Run下的Edit Configurations
配置Configuration分页中的Program arguments选项，就可以在idea中传入参数，应用自后，这样按住control+shift+r就可以启动运行了。
小窍门：如在工作时经常要写System.out.println(); 这个在idea里已经预设好了，直接输入sout 然后按tab键就可以了，你也可以试试输入soutm,soutp加tab键
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：50108次
排名：千里之外