spark使用文档

1.spark的安装和部署运行模式 本地模式 Standalone模式 Mesoes模式 yarn模式1.1. 本地模式安装jdk 安装scalatar -zxf scala-2.11.6.tgzspark版本只到 1.6.1，对应scala版本为 2.10.x配置文件/etc/profileexportJAVA_HOME= /usr/lib/jvm/java-1.7.0-openjdk-5.x86_64/jreexportSCALA_HOME=/home/yy/scalaexportSPARK_HOME=/home/yy/spark-1.2.0-bin-hadoop2.4exportPATH=$PATH:$JAVA_HOME/bin:$SCALA_HOME/bin:$SPARK_HOME/bin:$SPARK_HOME/sbinconf/spark-env.sh配置- 调试启动方式sh -x ./sbin/start-master.shSPARK_MASTER_IP=12SPARK_MASTER_PORT=7077SPARK_WORKER_CORES=3export MASTER=spark:/$SPARK_MASTER_IP:$SPARK_MASTER_PORTexport SPARK_MASTER_IP=12export SPARK_MASTER_PORT=7077export SPARK_WORKER_CORES=1export SPARK_WORKER_INSTANCES=1export SPARK_WORKER_MEMORY=1Gexport SPARK_WORKER_PORT=8888export MASTER=spark:/$SPARK_MASTER_IP:$SPARK_MASTER_PORT修改 conf/slaves.template 为conf/slaves启动master/opt/soft/spark/sbin/start-master.sh本地模式运行例子/opt/soft/spark/bin/spark-submit -master local* -class org.apache.spark.examples.SparkPi /opt/soft/spark/lib/spark-examples-1.6.1-hadoop2.6.0.jar hadoop 查看spark上传的文件hdfs dfs -ls /user/appuser/.sparkStagingspark shell任务提交spark-submit -class com.zxq.iov.cloud.spark.job.TBoxStreamJob -master yarn-cluster -num-executors 5 -conf spark.jars.excludes=org.apache.kafka:kafka_2.10 -jars hdfs:/appcluster/appjars/spark-hbase-1.0.0-SNAPSHOT.jar,hdfs:/appcluster/appjars/spark-kafka-1.0.0-SNAPSHOT.jar hdfs:/appcluster/appjars/spark-tbox-1.0.0-SNAPSHOT.jar-num-executorsspark-submit -class com.zxq.iov.cloud.spark.job.TBoxStreamJob -master yarn-cluster -num-executors 9 -executor-memory 2G -conf spark.streaming.kafka.maxRatePerPartition=100 -files hdfs:/appcluster/appconf/perties hdfs:/appcluster/appjars/spark-tbox-1.0.0-SNAPSHOT.jar部署中遇到的bugbug1启动master时报错failed to launch org.apache.spark.deploy.master.Master: at .InetAddress.getLocalHost(InetAddress.java:1469)windows下直接执行spark-shell报错Could not locate executable nullbinwinutils.exe初步怀疑是hadoop没有安装导致部署：Git库：Cloude/spark打的jar包就在target根目录下目前需要打jar包的模块有sp-as-sparksp-bda-sparkspark-tboxtrace-spark目标路径为hdfs:/appcluster/appjars/dev所涉及的服务器：13, 14, 15, 23qa所涉及的服务器：8, 9, 0, 1spark的jar包由hadoop进行分发管理SPARK与kafka整合kafka与现在的机制 可以多个executor对应一个partition不重复消费吗？已读多出来的executor要么是浪费不消费；要么是和其他executor轮询着消费；不会重复日志和bug定位spark on yarn web查看任务执行的jobyarn logs -applicationId application_1490703610049_0048job执行的错误日志yarn 的资源管理器界面进入SPARK版本spark实时计算程序优化1. spark流计算过程和 保存进hbase两个过程是否要程序人工处理成异步(异步衔接)第一个RDD实时计算，第二个RDD保存数据到hbase，如果habase出现性能问题阻塞，是否会导致spark性能下降,第一个RDD是否会阻塞如果改成队列方式，当队列满的时，再阻塞spark计算线程?cdh目前已有监控线程和性能的工具（针对spark rdd）spark context初始化val sc = new SparkContext(local,wordcount,conf)本地运行管理界面从yarnuid跳转到运行Job的spark-ui界面（spark job超链接在右侧，需要拖拽滚动条才能显现）查看job运行状况，是否有阻塞等查看执行任务中每个executor执行情况在stage tab中查看spark多线程场景和需求有一个sparkstreaming程序 程序里面有一些写hbase和es的逻辑，目前的逻辑是一条一条数据写，这样写效率很低，能不能在worker里面实现创建好两个线程，一个负责写hbase，一个负责写es.spark逻辑是负责一些计算，计算好以后扔给这两个线程就返回。由这两个线程异步批量写数据.现在就有一个问题了，如何在worker里面实现创建这样的线程可以在所有算子里面创建线程。比如map函数，foreachPartition函数里。然后通过单例的线程管理器去管理线程的异步回调，把结果输出到外部数据里。单例在每个worker实例（即JVM实例）独立的如何保证在退出进程的时候，线程的数据已经全部写到了hbase/es里面自己实现等所有异步线程完成后才通知Driver进行context.stop，否则会销毁所有Worker线程，异步线程的工作也毁于一旦如果在每个foreach算子里创建HBaseConnection，光是创建和释放就占用了70%以上的时间。应改成foreachPartition并每个记录复用Connection，效果立竿见影性能优化spark 一个JOB（stage）中的各个task之间是串行的每个 Sparkstage中包含若干个task，每个task串行地处理数据。在调试 Spark 的job时，task的个数可能是决定程序性能的最重要的参数。并行度指定在一个Spark程序的开始部分，有好多是用sparkContext的parallelize制作RDD的，是ParallelCollectionRDD，创建一个并行集合。例如sc.parallelize(0until numMappers, numMappers)创建并行集合的一个重要参数，是slices的数目（例子中是numMappers），它指定了将数据集切分为几份。在集群模式中，Spark将会在一份slice上起一个Task。典型的，你可以在集群中的每个cpu上，起2-4个Slice （也就是每个cpu分配2-4个Task）。一般来说，Spark会尝试根据集群的状况，来自动设定slices的数目。当让，也可以手动的设置它，通过parallelize方法的第二个参数。Spark主要提供了两种函数：parallelize和makeRDD：区别：A）makeRDD函数比parallelize函数多提供了数据的位置信息。B）两者的返回值都是ParallelCollectionRDD，但parallelize函数可以自己指定分区的数量，而makeRDD函数固定为seq参数的size大小。cache、checkpoint、broadcastcache如果不cache，rdd对象何时回回收如果不cache，当前foreach就销毁回收？简单代码 验证，创建一个spark-stream job，duration时间为较长时间，对rdd不进行缓存处理，查看jvm内存是否会溢出哪些RDD可以cache处理cache之后的生命周期，cache之后数据的可用范围Spark不支持ShuffleMapRDD的cache 我们经常会对RDD执行一系列Transformation算子操作，逻辑上每经历一次变换，就会将RDD转换为一个新的RDD，RDD会被划分成很多的分区分布到集群的多个节点中。 分区是逻辑概念，为了防止函数式数据不可变行（immutable）导致的内存需求无限扩张，使得系统内存被快速用完，Spark使用延迟执行（lazy）的方式执行，即只有操作累计到Action（行动），算子才会触发整个操作序列的执行，中间结果不会单独再重新分配内存，而是在同一个数据块上进行流水线操作。也就是说变换前后的新旧RDD的分区在物理上可能是同一块内存存储，这是Spark内部做的优化。有些RDD是计算的中间结果，其分区并不一定有相对应的内存或磁盘数据与之对应，所以如果想要复用某一个RDD，需要通过Cache算子，将数据缓存（或者说固化）到内存中checkpoint相关名词cache checkpoint broadcastBlockManagerMasterBlockManagerExecutorDrivershuttlebroadcasttransformationsaction第三方框架集成YDBInputDStream 时间窗、处理结果数据rdd传送stages for all jobTask（任务）：单个分区数据及上的最小处理流程单元。TaskSet（任务集）：由一组关联的，但互相之间没有Shuffle依赖关系的任务所组成的任务集。Stage（调度阶段）：一个任务集对应的调度阶段。Job（作业）：有一个RDD Action生成的一个或多个调度阶段所组成的一次计算作业。Application（应用程序）：Spark应用程序，由一个或多个作业组成。time window有状态转化操作（重点）时间窗口 规约函数逆函数SPARK API 及使用解读spark-streaming开发自定义streaming，参考spark-branch-1.6中的spark-streaming实现搜索关键词 onStart() onStop()pom中添加依赖org.apache.sparkspark-streaming_2.10参考 flume-streaming 实现a. 先跑通flume-streamingb. 参考flume-streaming 修改代码 跑最简单democ. 增加sftp处理逻辑socket-streamingnc启动客户端spark-streaming运行流程driver端运行，partition中运行Transformation和Action区别def mapU: ClassTag(f: T = U): RDDU = withScope val cleanF = sc.clean(f) new MapPartitionsRDDU, T(this, (context, pid, iter) = iter.map(cleanF)排序必须有黄色部分，排序才生效val lines = ssc.receiverStream(new SftpStreamingReceiver(host,userName,userPwd,port,path,timeOut,scanInterval)lines.foreachRDD(rdd = val middle = rdd.distinct().map(t=(t.split(,)(0),t.split(,)(1).sortBy(x = x._1,true) middle.collect().foreach(str=println(strDistinct:+str)IDEA中使用IDEA中提交任务VM options: -verbose:class -Dspark.master=local spark-streaming 返回对象为对象流？需确认的问题当前的spark程序在cdh集群存在问题，在其他环境(hadoop原生环境)是否没有问题参考url:总体介绍Spark Streaming Programming Guide(官方文档)/docs/latest/streaming-programming-guide.htmlspark简明介绍/infodetail-954616.htmlSpark运行架构/shishanyuan/p/4721326.htmlSpark中文手册4：Spark之基本概念（重要参考）/thread-11516-1-1.html部署虚拟机上搭建spark集群/wjcquking/article/details/45286755性能优化spark 中如何划分stage？/shadow_mi/article/details/51821613用实例说明Spark stage划分原理/bonelee/p/6039469.htmlApache Spark Jobs 性能调优（二）/xiaop1987/note/102894Spark 多个Stage执行是串行执行的么?/archives/62119spark JVM调优之调节executor堆外内存与连接等待时长/hutao_hadoop/article/details/52694550spark幂等操作Spark版本定制4-Spark Streaming事务处理彻底理解/thread-20750-1-1.html实际应用场景例子Kafka+Spark Streaming+Redis实时计算整合实践/archives/1097.html任务提交Spark提交任务到集群/caiandyong/article/details/51831434Spark-submit提交任务到集群/kinger0/article/details/46562473Spark优化：禁止应用程序将依赖的Jar包传到HDFS/articles/7rMRV3spark提交任务的流程/wnbahmbb/p/6266659.htmlspark-streamingSpark Streaming 自定义接收器 java实现/ouyang111222/article/details/50414621自定义Spark Streaming接收器(Receivers) scala实现/archives/1594.html任务计划Spark调度系列-5.Spark task和Stage的跳过执行(ui显示task和stage skipped)/u012684933/article/details/50378725探索Spark源码-在Spark中Job是并行执行的还是串行执行的？/thread-16502-1-1.html集成和调试使用IDEA运行Spark程序/lsshlsw/article/details/39668911Spark基础随笔：Spark1.6 Idea下远程调试/jiangpeng59/article/details/52444466spark job 独立的log4j配置方法/xueba207/article/details/50436684spark 与mogoDB整合MongoDB + Spark: 完整的大数据解决方案/MongoDB/31116.html快的mongodb使用架构/MongoDB/2016/0105/6816.html监控Spark Metrics配置详解/archives/1341.htmljmx监控spark executor配置/linux_ja/article/details/46520119spark与kafka集成spark streaming消费kafka消息的两种方式对比/articles/VvqQjqJsparkstreaming消费kafka中的数据/high2011/article/details/51585626Spark Streaming 自适应上游 kafka topic partition 数目变化/u011007180/article/details/53327235Spark Streaming 与 Kafka 集成分析/allwefantasy/article/details/50447700Michael G. Noll：整合Kafka到Spark Streaming代码示例和挑战/article/2014-12-31/2823384-kafka-spark-streaming-integration-example-tutorialdibbhatt/kafka-spark-consumer/dibbhatt/kafka-spark-consumer/blob/master/src/main/java/consumer/kafka/KafkaUtils.javaspark分析数据场景Spark入门学习(一) 分析nginx log文件/articles/BNVjArmspark与sftp整合Spark SFTP Connector Library/springml/spark-sftpspark-assembly包的编译CDH5.1.0编译spark-assembly包来支持hive/aaa1117a8w5s6d/article/details/44307207Hadoop2.6.0-cdh5.4.1源码编译安装/articles/vQzMRbQspark与redis集成/sundujing/article/details/51480085/art/201603/507663.htm性能优化spark 应用程序性能优化12 个优化方法http:/litaotao.github.io/boost-spark-application-performance排序Spark学习之旅（一）SortBy/sheyuedeye/article/details/43406179Transformspark transform系列_distinct/u014393917/article/details/50602431DataFrameSpark SQL 1.3.0 DataFrame介绍、使用及提供了些完整的数据写入/forum.php?mod=viewthread&tid=12358&page=1进化的Spark, 从DataFrame说起/articles/IzeY7zM有状态rdd及时间窗reduceByKeyAndWindow的使用/blog/2308138包含有时间窗内容/u013089961/article/details/52328127Spark Streaming在线动态计算特定时间窗口下热门商品排名/sd637/article/details/51517197SparkStreaming之窗口函数/legotime/article/details/51836040spark与drools整合相关名词：cep sparkStreaming和drools整合/net/201702/596979.html酷狗音乐的大数据实践（纯干货）spark(包括跟drools整合的内容)/nosql/1321044.htmcdh上对spark 与drools整合介绍/blog/2015/11/how-to-build-a-complex-event-processing-app-on-apache-spark-and-drools/shuffle详细探究Spark的shuffle实现/johnny_lee/article/details/22619585/article/2015-07-08/2825162Spark Streaming 流计算优化记录(6)-GC优化与shuffle service/butterluo/article/details/47084897Spark Shuffle的技术演进/p/4c5c2e535da5Spark Streaming 流计算优化记录(6)-GC优化与shuffle service/butterluo/article/details/47084897spark的作业调度问题/article/details?id=52670867Spark技术内幕：Shuffle的性能调优/anzhsoft/article/details/42837211Spark性能优化和shuffle搏斗/lib/view/open1463877974257.htmlIDEA中运行sparkspark (java API) 在Intellij IDEA中开发并运行/dream_an/article/details/54915894spark 在idea中运行/info-detail-1192175.html运行方式yarn-client 和 yarn-cluster差别/thread-332290-1-1.htmlspark-streaming日志和bug定位查看spark on yarn的日志和程序状态的方法(web UI)/high2011/article/details/52132646spark日志输出到本地,log4j的配置/stark_summer/article/details/46929481spark相关错误和解决方法/db/mysql/201611/211267.shtmlSpark Core 问题详解/developerworks/cn/opensource/os-cn-spark-core/Spark-submit提交任务到集群/kinger0/article/details/46562473spark多线程在spark job中创建线程/topics/392041481?page=1cache和persistspark中cache和persist的区别/houmou/article/details/52491419Spark Cache机制/blog/2182146Spark Streaming 数据清理机制/p/f068afb23c77CacheManager彻底解密：CacheManager运行原理流程图和源码详解/jcchoiling/p/6510969.htmlspark RDD缓存代码分析/archives/1532.htmlcheckpoint/p/00b591c5f623Spark Streaming 缓存、Checkpoint机制/lovehuangjiaju/article/details/50102831/spark rdd checkpoint的用法注意点/xiao_jun_0820/article/details/50475351yarnYarn的原理与资源调度/yunpiao123456/article/details/51802986spark不同模式下应用程序运行的日志存放位置/u011878191/article/details/45894167Spark应用程序运行的日志存在哪里/u011878191/article/details/45891079源码阅读搭建Spark源码阅读环境/zc741520/blog/499138并行度（execute-num¶lel）spark中的动态executor分配/u014393917/article/details/50630829jar包替换和升级重新编译、安装spark assembly，使CDH5.5.1支持sparkSQL/zhong_han_jun/article/details/50358745Spark技术内幕：Client，Master和Worker 通信源码解析/luccs624061082/article/detai