TA的每日心情 | 无聊 7 小时前 |
---|
签到天数: 528 天 连续签到: 1 天 [LV.9]测试副司令
|
1. 设置消息尺寸最大值
- def main(args: Array[String]) {
- System.setProperty("spark.akka.frameSize", "1024")
- }
复制代码
2.与yarn结合时设置队列
- val conf=new SparkConf().setAppName("WriteParquet")
- conf.set("spark.yarn.queue","wz111")
- val sc=new SparkContext(conf)
复制代码
3.运行时使用yarn分配资源,并设置--num-executors参数
- nohup /home/SASadm/spark-1.4.1-bin-hadoop2.4/bin/spark-submit
- --name mergePartition
- --class main.scala.week2.mergePartition
- --num-executors 30
- --master yarn
- mergePartition.jar >server.log 2>&1 &
复制代码
4.读取impala的parquet,对String串的处理
- sqlContext.setConf("spark.sql.parquet.binaryAsString","true")
复制代码
5.parquetfile的写
- case class ParquetFormat(usr_id:BigInt , install_ids:String )
- val appRdd=sc.textFile("hdfs://").map(_.split("\t")).map(r=>ParquetFormat(r(0).toLong,r(1)))
- sqlContext.createDataFrame(appRdd).repartition(1).write.parquet("hdfs://")
复制代码
6.parquetfile的读
- val parquetFile=sqlContext.read.parquet("hdfs://")
- parquetFile.registerTempTable("install_running")
- val data=sqlContext.sql("select user_id,install_ids from install_running")
- data.map(t=>"user_id:"+t(0)+" install_ids:"+t(1)).collect().foreach(println)
复制代码
7.写文件时,将所有结果汇集到一个文件
8.如果重复使用的rdd,使用cache缓存
9.spark-shell 添加依赖包
- spark-1.4.1-bin-hadoop2.4/bin/spark-shell local[4] --jars code.jar
复制代码
10.spark-shell使用yarn模式,并使用队列
- spark-1.4.1-bin-hadoop2.4/bin/spark-shell --master yarn-client --queue wz111
复制代码
|
|