校园网站建设情况统计表/外呼系统电销

1 介绍

在spark中，支持4种运行模式：

• local：开发时使用
• standalone：spark自带，如果一个集群时standalone的话，就需要在多台机器上同时部署spark环境。
• yarn：建议在生产上使用该模式，统一使用yarn进行整个集群作业（MR，spark）的资源调度。
• mesos

注意：

• 不管使用什么模式，spark应用程序的代码是一模一样的
• spark支持可插拔的集群管理模式
• 对于yarn而言，spark application仅仅是一个客户端而已

1）yarn之client模式

• driver运行在client端（提交spark作业的机器）
• client会和请求到的container进行通信来完成作业的调度和执行，client是不能退出的
• 日志信息会在控制台输出，便于我们测试
• application master 负责资源的申请

2）yarn之cluster模式

• driver运行在application master中。
• client只要提交完作业之后就可以关掉，因为作业已经在yarn上运行了。
• 日志在终端是看不见的，因为日志在driver上，只能通过yarn logs -applicationId application_id查看。
• application master：资源申请与任务调度。

2 程序

1）修改pom.xml文件

<!-- 以下依赖不需要打进jar包 添加<scope> -->
<!-- scala -->
<dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version><scope>provided</scope>
</dependency><!-- sparksql -->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>${spark.version}</version><scope>provided</scope>
</dependency><!-- 使用hiveContext需要的依赖-->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.11</artifactId><version>${spark.version}</version><scope>provided</scope>
</dependency><!-- jdbc 操作依赖 使用thriftserver-->
<dependency><groupId>org.spark-project.hive</groupId><artifactId>hive-jdbc</artifactId><version>1.2.1.spark2</version><scope>provided</scope>
</dependency>

<!-- 添加plugin，将需要的依赖一起打包-->
<plugin><artifactId>maven-assembly-plugin</artifactId><configuration><archive><manifest><mainClass></mainClass></manifest></archive><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration>
</plugin>

2）SparkStatCleanJobYarn.scala

package logimport org.apache.spark.sql.{SaveMode, SparkSession}/*** spark清洗操作运行在yarn上*/
object SparkStatCleanJobYarn {def main(args: Array[String]): Unit = {if (args.length!=2) {println("usage: sparkstatcleanjobyarn <inputpath> <outputpath>")System.exit(1)}val Array(inputPath,outputPath)=argsval spark=SparkSession.builder().getOrCreate()val accessRDD=spark.sparkContext.textFile(inputPath)val accessDF = spark.createDataFrame(accessRDD.map(x => AccessConvertUtil.parseLog(x)),AccessConvertUtil.struct)accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save(outputPath)spark.stop()}
}

3）TopNStatJobYARN.scala，更改TopNStatJob.scala中的main函数

def main(args: Array[String]): Unit = {if (args.length != 2) {println("usage: topnstatjobyarn <inputpath> <day>")System.exit(1)}val Array(inputPath, day) = argsval spark = SparkSession.builder().config("spark.sql.sources.partitionColumnTypeInference.enabled", "false").getOrCreate()val accessDF = spark.read.format("parquet").load(inputPath)StatDAO.deleteData(day)// 最受欢迎的topN课程videoAccessTopNStat(spark, accessDF, day)// 按地市统计topN课程cityAccessTopNStat(spark, accessDF, day)// 按流量统计topN课程videoTrafficsTopNStat(spark, accessDF, day)spark.stop()}

3 提交运行

1）项目打包

mvn assembly:assembly

2）通过 spark-submit方式提交

export HADOOP_CONF_DIR=/Users/Mac/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop/

// 提交清洗数据作业
spark-submit \
--class log.SparkStatCleanJobYarn \
--name SparkStatCleanJobYarn \
--master yarn \
--executor-memory 1G \
--num-executors 1 \
--files /Users/Mac/pro-use-file/ipDatabase.csv,/Users/Mac/pro-use-file/ipRegion.xlsx \
/Users/Mac/my-lib/sql-1.0-jar-with-dependencies.jar \
hdfs://localhost:8020/access.log hdfs://localhost:8020/spark-clean-out/

// 提交统计作业
spark-submit \
--class log.TopNStatJobYARN \
--name TopNStatJobYARN \
--master yarn \
--deploy-mode client \
--executor-memory 1G \
--num-executors 1 \
/Users/Mac/my-lib/sql-1.0-jar-with-dependencies.jar \
hdfs://localhost:8020/spark-clean-out/ 20161110

3）结果

• 清洗数据作业可以在HDFS上的输出目录中找到输出文件。
• 统计作业可以到数据库中查看输出数据。