本文主要介绍spark join相关操作,Java描述。
讲述三个方法spark join,left-outer-join,right-outer-join
我们以实例来进行说明。我的实现步骤记录如下。
1、数据准备
2、HSQL描述
3、Spark描述
1、数据准备
我们准备两张Hive表,分别是orders(订单表)和drivers(司机表),通过driver_id字段进行关联。数据如下:
orders
hive (gulfstream_test)> select * from orders; OK orders.order_id orders.driver_id Time taken: row(s)
drivers
hive (gulfstream_test)> select * from drivers; OK drivers.driver_id drivers.car_id Time taken: row(s)
2、HSQL描述
JOIN
自然连接,输出连接键匹配的记录。
hive (gulfstream_test)> select * from orders t1 join drivers t2 on (t1.driver_id = t2.driver_id) ; OK t1.order_id t1.driver_id t2.driver_id t2.car_id Time taken: row(s)
LEFT OUTER JOIN
左外链接,输出连接键匹配的记录,左侧的表无论匹配与否都输出。
hive (gulfstream_test)> select * from orders t1 left outer join drivers t2 on (t1.driver_id = t2.driver_id) ; OK t1.order_id t1.driver_id t2.driver_id t2.car_id NULL NULL NULL NULL Time taken: row(s)
RIGHT OUTER JOIN
右外连接,输出连接键匹配的记录,右侧的表无论匹配与否都输出。
hive (gulfstream_test)> select * from orders t1 right outer join drivers t2 on (t1.driver_id = t2.driver_id) ; OK t1.order_id t1.driver_id t2.driver_id t2.car_id Time taken: row(s)
3、Spark描述
Join.java
spark实现join的方式也是通过RDD的算子,spark同样提供了三个算子join,leftOuterJoin,rightOuterJoin。
在下面给出的例子中,我们通过spark-hive读取了Hive表中的数据,并将DataFrame转化成了RDD。
在join之后,通过collect()函数把数据拉到Driver端本地,并通过标准输出打印。
需要指出的是:
1)join算子(join,leftOuterJoin,rightOuterJoin)只能通过PairRDD使用;
2)join算子操作的Tuple2<Object1, Object2>类型中,Object1是连接键,我只试过Integer和String,Object2比较灵活,甚至可以是整个Row。
package com.kangaroo.studio.algorithms.join; import com.google.common.base.Optional; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.sql.DataFrame; import org.apache.spark.sql.Row; import org.apache.spark.sql.hive.HiveContext; import scala.Tuple2; import java.io.Serializable; import java.util.Iterator; /* * spark-submit --queue=root.zhiliangbu_prod_datamonitor spark-join-1.0-SNAPSHOT-jar-with-dependencies.jar * */ public class Join implements Serializable { private transient JavaSparkContext javaSparkContext; private transient HiveContext hiveContext; /* * 初始化Load * 创建sparkContext, sqlContext, hiveContext * */ public Join() { initSparckContext(); initHiveContext(); } /* * 创建sparkContext * */ private void initSparckContext() { String warehouseLocation = System.getProperty("user.dir"); SparkConf sparkConf = new SparkConf() .setAppName("spark-join") .set("spark.sql.warehouse.dir", warehouseLocation) .setMaster("yarn-client"); javaSparkContext = new JavaSparkContext(sparkConf); } /* * 创建hiveContext * 用于读取Hive中的数据 * */ private void initHiveContext() { hiveContext = new HiveContext(javaSparkContext); } public void join() { /* * 生成rdd1 * */ String query1 = "select * from gulfstream_test.orders"; DataFrame rows1 = hiveContext.sql(query1).select("order_id", "driver_id"); JavaPairRDD<String, String> rdd1 = rows1.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() { @Override public Tuple2<String, String> call(Row row) throws Exception { String orderId = (String)row.get(0); String driverId = (String)row.get(1); return new Tuple2<String, String>(driverId, orderId); } }); /* * 生成rdd2 * */ String query2 = "select * from gulfstream_test.drivers"; DataFrame rows2 = hiveContext.sql(query2).select("driver_id", "car_id"); JavaPairRDD<String, String> rdd2 = rows2.toJavaRDD().mapToPair(new PairFunction<Row, String, String>() { @Override public Tuple2<String, String> call(Row row) throws Exception { String driverId = (String)row.get(0); String carId = (String)row.get(1); return new Tuple2<String, String>(driverId, carId); } }); /* * join * */ System.out.println(" ****************** join *******************"); JavaPairRDD<String, Tuple2<String, String>> joinRdd = rdd1.join(rdd2); Iterator<Tuple2<String, Tuple2<String, String>>> it1 = joinRdd.collect().iterator(); while (it1.hasNext()) { Tuple2<String, Tuple2<String, String>> item = it1.next(); System.out.println("driver_id:" + item._1 + ", order_id:" + item._2._1 + ", car_id:" + item._2._2 ); } /* * leftOuterJoin * */ System.out.println(" ****************** leftOuterJoin *******************"); JavaPairRDD<String, Tuple2<String, Optional<String>>> leftOuterJoinRdd = rdd1.leftOuterJoin(rdd2); Iterator<Tuple2<String, Tuple2<String, Optional<String>>>> it2 = leftOuterJoinRdd.collect().iterator(); while (it2.hasNext()) { Tuple2<String, Tuple2<String, Optional<String>>> item = it2.next(); System.out.println("driver_id:" + item._1 + ", order_id:" + item._2._1 + ", car_id:" + item._2._2 ); } /* * rightOuterJoin * */ System.out.println(" ****************** rightOuterJoin *******************"); JavaPairRDD<String, Tuple2<Optional<String>, String>> rightOuterJoinRdd = rdd1.rightOuterJoin(rdd2); Iterator<Tuple2<String, Tuple2<Optional<String>, String>>> it3 = rightOuterJoinRdd.collect().iterator(); while (it3.hasNext()) { Tuple2<String, Tuple2<Optional<String>, String>> item = it3.next(); System.out.println("driver_id:" + item._1 + ", order_id:" + item._2._1 + ", car_id:" + item._2._2 ); } } public static void main(String[] args) { Join sj = new Join(); sj.join(); } }
pom.xml
pom依赖
这里只依赖spark-core和spark-hive两个jar。
<dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.10</artifactId> <version>1.6.0</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.10</artifactId> <version>1.6.0</version> <scope>provided</scope> </dependency> </dependencies>
打包方式
<build> <plugins> <plugin> <artifactId>maven-assembly-plugin</artifactId> <configuration> <archive> <manifest> <!--这里要替换成jar包main方法所在类 --> <mainClass>com.kangaroo.studio.algorithms.join.Join</mainClass> </manifest> </archive> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <!-- this is used for inheritance merges --> <phase>package</phase> <!-- 指定在打包节点执行jar包合并操作 --> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>1.6</source> <target>1.6</target> </configuration> </plugin> </plugins> </build>
执行结果
其中Optional.absent()表示的就是null,可以看到和HSQL是一致的。
Application ID is application_1508228032068_2746260, trackingURL: http://10.93.21.21:4040 ****************** join ******************* driver_id:, order_id:, car_id: ****************** leftOuterJoin ******************* driver_id:, order_id:, car_id:Optional.absent() driver_id:, order_id:, car_id:Optional.absent() driver_id:, order_id:, car_id:Optional.of() ****************** rightOuterJoin ******************* driver_id:, order_id:Optional.absent(), car_id: driver_id:, order_id:Optional.of(), car_id: