spark join操作
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了spark join操作相关的知识，希望对你有一定的参考价值。
本文主要介绍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 1000 5000 1001 5001 1002 5002 Time taken: 0.387 seconds, Fetched: 3 row(s)
drivers
hive (gulfstream_test)> select * from drivers; OK drivers.driver_id drivers.car_id 5000 100 5003 103 Time taken: 0.036 seconds, Fetched: 2 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 1000 5000 5000 100 Time taken: 36.079 seconds, Fetched: 1 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 1000 5000 5000 100 1001 5001 NULL NULL 1002 5002 NULL NULL Time taken: 36.063 seconds, Fetched: 3 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 1000 5000 5000 100 NULL NULL 5003 103 Time taken: 30.089 seconds, Fetched: 2 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:5000, order_id:1000, car_id:100 ****************** leftOuterJoin ******************* driver_id:5001, order_id:1001, car_id:Optional.absent() driver_id:5002, order_id:1002, car_id:Optional.absent() driver_id:5000, order_id:1000, car_id:Optional.of(100) ****************** rightOuterJoin ******************* driver_id:5003, order_id:Optional.absent(), car_id:103 driver_id:5000, order_id:Optional.of(1000), car_id:100
以上是关于spark join操作的主要内容，如果未能解决你的问题，请参考以下文章