首先配置jupyter config文件。
jupyter-notebook --generate-config
修改jupyter config文件
c.NotebookApp.port = 18888
c.NotebookApp.ip = '0.0.0.0'
c.NotebookApp.allow_root = True
当然要配置好spark，emr环境spark已经完全配置正确。配置pyspark参数
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook'
启动pyspark即可。
pyspark --master yarn
最近转战Spark相关领域，编写了spark版本的LDA算法，在此过程中对程序不断调优同时也对spark系统相关参数进行调优，得到了一下一些经验。
首先是对于RDD转化为PariRDD，初学者可能会有很多疑问，为什么通过map将RDD转化为tuple后应该有相应的PairRDD操作，但是IDE并么有提示，原因是缺少隐式转换，加入 import org.apache.spark.SparkContext._ 即可解决。
其次，有些情况下combineByKey的性能是十分好的，相当于在每个paritition处做combiner，减少网络传输和处理时间。
下面是一些参数调整：
1、spark.driver.maxResultSize 8g driver获得处理结果的最大内存数，由于我要处理大矩阵，所以这个参数还是不得不改的
2、spark.yarn.executor.memoryOverhead 2048 跑了一段时间后发现很多executor堆外内存占用过大，采用这个参数后稍好
3、spark.shuffle.blockTransferService nio spark 1.2.0以后shuffle service改为了netty，这个很扯淡，我改为nio后堆外内存较少了很多，同时处理时间提示提升了一倍
先写到这儿，感谢同事何良均提供第三个参数，他的blog地址是http://blog.csdn.net/hit_hlj_sgy/article/list/2
我的spark LDA github地址是https://github.com/jiangyu/SparkLDA
Spark编译不是很复杂，使用maven就可以编译，但对maven的内存要求较高。
1、export MAVEN_OPTS=”-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512m”
2、mvn -Pyarn -Phadoop-2.4 -Dhadoop.version=2.4.0 -DskipTests clean package
因为我是要跑在yarn上，其中要指定hadoop版本。
除了第一次需要下载大量jar包，其它编译大概需要20分钟左右，编译好后在$SPARK_HOME/assembly/target/scala-2.10/ 下面有个jar包，大概100多兆，这是我们部署用的。
Spark部署和配置也不难，从官网下载发布包(非刚才的源码包)即可,需要以下几步就可以部署使用。
1、$SPARK_HOME/lib 下替换spark-assmbly的jar包，替换为刚才编译出来的jar包
2、在$SPARK_HOME/conf 配置spark-env.sh
export SPARK_YARN_JAR==hdfs://ns1/spark/jar/spark-assembly-1.2.1-hadoop2.4.0.jar #上传刚编译的jar包到dfs上，让所有worked都能get到
export HADOOP_CONF_DIR=/usr/local/hadoop-2.4.0/etc/hadoop #Hadoop配置路径
export YARN_CONF_DIR=$HADOOP_CONF_DIR #Yarn配置路径
export SPARK_DAEMON_JAVA_OPTS=”-Djava.library.path=/usr/local/hadoop-2.4.0/lib/native/ -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/data0/hadoop/gclog/heap.dump.spark”
export SPARK_SUBMIT_LIBRARY_PATH=/usr/local/hadoop-2.4.0/lib/native/ #LD_LIBRARY使用，调用native库用，如果使用lzo等配置好
3、在$SPARK_HOME/conf下配置spark-default.sh
spark.yarn.historyServer.address *.*.*.*:18080 #yarn historyserver地址
spark.eventLog.enabled true
spark.eventLog.dir hdfs://ns1/spark/logs
至此，Spark配置完成，如果在YARN上跑一下测试可以用$SPARK_HOME/bin/spark-shell —master yarn-client
最近一直在看Spark以及Hive相关的东西，Spark SQL就是使用Hive SQL Parser将hive语句转化为RDD,方便Hive用户平滑迁移。
首先就是如何打包：
1、从github上下载源码，地址是https://github.com/apache/spark，我选取的是1.0.2(1.1.0在提交时候出现Permission denied问题，我放弃了)
2、安装maven,scala等，Hadoop集群，并设定home目录
3、通过Maven打包前修改Maven的jvm参数，export MAVEN_OPTS=”-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512”
4、根据HADOOP版本通过MAVEN打包,比如对2.4版本 bash make-distribution.sh –name spark-1.0.2 –tgz -Pyarn -Phadoop-2.4 -Dhadoop.version=2.4.0 打包出来是一个tgz的包，上传到部署的机器上 部署问题：
5、解压tgz的包，将hive需要的mysql-jdbc的jar包拷贝到lib下，同时将hive配置文件hive-site.xml配置好拷贝到conf下
6、修改配置文件，修改spark-default.conf
spark.yarn.historyServer.address 10.39.5.23:18080 // spark history server position spark.eventLog.enabled true spark.eventLog.dir hdfs://ns1/user/jiangyu2/logs |
修改spark-env.sh
export SPARK_JAR=hdfs://ns1/user/jiangyu2/spark-assembly-1.1.0-SNAPSHOT-hadoop2.4.0.jar export HADOOP_CONF_DIR=/usr/local/hadoop-2.4.0/etc/hadoop export YARN_CONF_DIR=/usr/local/hadoop-2.4.0/etc/hadoop export SPARK_YARN_USER_ENV="CLASSPATH=/usr/local/hadoop-2.4.0/etc/hadoop/" export SPARK_SUBMIT_LIBRARY_PATH=/usr/local/hadoop-2.4.0/lib/native/ export SPARK_CLASSPATH=$SPARK_CLASSPATH |
7、启动spark shell
$ ./bin/spark-shell --master yarn-client scala> val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.input.dir.recursive is deprecated. Instead, use mapreduce.input.fileinputformat.input.dir.recursive 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.max.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.maxsize 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.min.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.min.split.size.per.rack is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize.per.rack 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.min.split.size.per.node is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize.per.node 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.reduce.tasks is deprecated. Instead, use mapreduce.job.reduces 14/09/17 17:57:07 INFO Configuration.deprecation: mapred.reduce.tasks.speculative.execution is deprecated. Instead, use mapreduce.reduce.speculative hiveContext: org.apache.spark.sql.hive.HiveContext = org.apache.spark.sql.hive.HiveContext@1f3b0168 scala> import hiveContext._ import hiveContext._ scala> hql("SELECT t1.fans_uid,t3.user_type FROM (SELECT fans_uid,atten_uid,time FROM ods_user_fanslist WHERE dt = '20140909') t1 JOIN (SELECT uid,user_type_id AS user_type,user_status,reg_time FROM mds_user_info WHERE dt = '20140909') t3 ON t1.atten_uid = t3.uid limit 10”).collect().foreach(println) |
8、写java程序提交同样的hql
package nimei; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.sql.hive.api.java.JavaHiveContext; public class NiDaYe { public static void main(String[] args) { SparkConf sparkConf = new SparkConf().setAppName("caocaocao"); JavaSparkContext ctx = new JavaSparkContext(sparkConf); JavaHiveContext hiveCtx = new org.apache.spark.sql.hive.api.java.JavaHiveContext(ctx); // Queries are expressed in HiveQL. hiveCtx.hql("SELECT count(fans_uid),user_status from(SELECT fans_uid,atten_uid,time FROM ods_user_fanslist WHERE dt = '20140909') " + "t1 JOIN (SELECT uid,user_type_id AS user_type,user_status,reg_time FROM " + " mds_user_info WHERE dt = '20140909') t3 ON t1.atten_uid = t3.uid group by user_status").collect(); } } |
最后就是applicationMaster页面了 