纵有疾风起
人生不言弃

Spark第一个应用程序

首先要对源码进行编译,生成对应hadoop版本的spark开发程序jar包,上篇已经写了具体的过程,这里不再赘述。

在安装spark的机器上,下载eclipse-java-x86_64版本,将spark-assembly.jar和spark/lib下全部加进路径,建立普通java project

WordCount代码

package sparktest.util.test;import java.util.Arrays;import java.util.regex.Pattern;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.FlatMapFunction;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import scala.Tuple2;public final class JavaWordCount {  private static final Pattern SPACE = Pattern.compile(" ");  public static void main(String[] args) throws Exception {    if (args.length < 2) {      System.err.println("Usage: JavaWordCount <master> <file>");      System.exit(1);    }    JavaSparkContext ctx = new JavaSparkContext(args[0], "JavaWordCount",        System.getenv("SPARK_HOME"), JavaSparkContext.jarOfClass(JavaWordCount.class));    ctx.addJar("/opt/eclipse/JavaSparkT.jar");    JavaRDD<String> lines = ctx.textFile(args[1], 1);        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {      @Override      public Iterable<String> call(String s) {        return Arrays.asList(SPACE.split(s));      }    });        JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {      @Override      public Tuple2<String, Integer> call(String s) {        return new Tuple2<String, Integer>(s, 1);      }    });        JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {      @Override      public Integer call(Integer i1, Integer i2) {        return i1 + i2;      }    });    counts.saveAsTextFile(args[2]); //    counts.s    /*List<Tuple2<String, Integer>> output = counts.collect();    for (Tuple2<?,?> tuple : output) {      System.out.println(tuple._1() + ": " + tuple._2());    }*/    System.exit(0);  }}

 

输入的三个参数:

spark://master:7077 /usr/local/hadoop/hadoop-2.4.0/test.txt /usr/local/hadoop/hadoop-2.4.0/result.txt

分别spark入口 本地输入文件 本地输出文件

输出:

[root@localhost result.txt]# cat part-00000 (xing,2)(xiao,1)(ya,2)(shi,1)(,2)(wo,1)(yi,4)(zhi,1)

 

注意: ctx.addJar(“/opt/eclipse/JavaSparkT.jar”);  这一句很关键,要不然会报错

问题:

14/07/07 10:26:11 WARN TaskSetManager: Lost TID 0 (task 1.0:0)
14/07/07 10:26:11 WARN TaskSetManager: Loss was due to java.lang.ClassNotFoundException
java.lang.ClassNotFoundException: JavaWordCount$1
    at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:355)

解决办法:

spark在运行的时候需要将类分发到各个节点上去,然后在各个节点调用这个类去完成任务,在java程序里边需要将这个类本身提交到各个节点上去

可以通过java普通的方法调用的方式实现或者 在eclipse里边通过右键run as java application的方式运行。不过需要在程序中用addJar()方法或者在初始化SparkContext的时候将本身的WordCount 的Jar包增加进去。

 

Spark类型:

Supported types for Avro -> SparkSQL conversion

avro types to SparkSQL types:

boolean -> BooleanTypeint -> IntegerTypelong -> LongTypefloat -> FloatTypedouble -> DoubleTypebytes -> BinaryTypestring -> StringTyperecord -> StructTypeenum -> StringTypearray -> ArrayTypemap -> MapTypefixed -> BinaryType

Spark Load/Save 方式

 1.3.0统一了load/save API,让用户按需自由选择外部数据源。这套API包括:

  1.SQLContext.table

  从SQL表中加载DataFrame。

  2.SQLContext.load

  从指定的外部数据源加载DataFrame。

  3.SQLContext.createExternalTable

  将指定位置的数据保存为外部SQL表,元信息存入Hive metastore,并返回包含相应数据的DataFrame。

  4.DataFrame.save

  将DataFrame写入指定的外部数据源。

  5.DataFrame.saveAsTable

  将DataFrame保存为SQL表,元信息存入Hive metastore,同时将数据写入指定位置。

 

Spark与Hive的兼容特性

  • 与hive兼容

    sparkSQL可以与hive metastore,serdes和UDF兼容。sparkSQL thrift jdbc服务能够与当前已经安装的hive进行兼容,不需要修改已存在的hive metastore或者改变数据存放目录。

    支持的hive特性

    sparkSQL支持很多hive特性,如:

  • hive 查询语句:select,group by,order by,cluster by,sort by
  • 所有hive操作:关系型操作(=, ⇔, ==, <>, <, >, >=, <=等),算术操作(+, -, *, /, %等),裸机操作(AND, &&, OR, ||等),复杂类型构造,数学函数(sign, ln, cos等),字符串函数(instr, length, printf等)
  • 用户定义函数(UDF)
  • 用户定义聚合函数(UDAF)
  • 用户定义序列化格式(serdes)
  • join:join,{left|right|full} outer join,left semi join,cross join
  • 联合查询
  • 子查询:select col from(select a+b as col from t1)t2
  • 取样操作
  • 解释操作
  • 表分割
  • 所有的hive DDL函数:create table,create table as select,alter table
  • 大部分的hive数据类型:TINYINT,SMALLINT,INT,BIGINT,BOOLEAN,FLOAT,DOUBLE,STRING,BINARY,TIMESTAMP,ARRAY<>,MAP<>,STRUCT<> 

    不支持的hive功能

    主要的hive特性 
  • sparkSQL目前不支持使用动态分片向表插入数据
  • sparkSQL不支持带有桶的表,桶是hive表分片的hash分片方式
    深奥的hive特性 
  • hive中带有分片的表使用不同的输入格式,在sparkSQL中,所有的表分片使用相同的输入格式
  • hive支持在join操作使用non-equi的条件(如key<10),在sparkSQL中如果使用这种方式会输出错误的结果
  • UNION和DATE类型
  • 唯一join
  • 单查询多插入
  • sparkSQL不支持piggyback浏览来收集列统计,只支持操作hive metastore中的sizeInBytes字段
    hive输入输出格式
  • sparkSQL只支持TextOutputFormat
  • hadoop归档文件
    hive优化

    很多的hive优化目前都不能用户sparkSQL,其中一些(如索引)对于sparkSQL不是很重要的,因为sparkSQL是内存计算模型,其他的一些会在未来的sparkSQL版本中得到支持:

  • 块级别的bitmap索引和虚拟字段(用来建立索引)
  • 自动将join转换为map join:在大表与很多小表进行join时,hive会自动将join转换为map join,下个版本的sparkSQL会得到支持
  • 自动决定join和groupby时reducer的数量,目前,sparkSQL需要使用SET spark.sql.shuffle.partitions=[num_tasks];来控制并行度
  • 只查询元数据:hive在查询时可以只查询元数据,sparkSQL需要部署任务来计算结果
  • sparkSQL不支持hive中的skew data flag
  • sparkSQL不支持hive的STREAMTABLE
  • 合并多个小文件作为查询结果:如果查询结果包括很多的小文件,hive可以合并这些小文件为大文件避免HDFS元数据容量溢出。sparkSQL暂时不支持该特性
 

文章转载于:https://www.cnblogs.com/kxdblog/p/4503525.html

原著是一个有趣的人,若有侵权,请通知删除

未经允许不得转载:起风网 » Spark第一个应用程序

分享到: 生成海报

评论 抢沙发

评论前必须登录!

立即登录   注册

切换注册

登录

忘记密码 ?

切换登录

注册

我们将发送一封验证邮件至你的邮箱, 请正确填写以完成账号注册和激活