Dataset<Row> resultDs = dsParsed
.groupBy("enodeb_id", "ecell_id")
.agg(functions.first("scan_start_time").alias("scan_start_time1"),functions.first("insert_time").alias("insert_time1"),functions.first("mr_type").alias("mr_type1"),functions.first("mr_ltescphr").alias("mr_ltescphr1"),functions.first("mr_ltescpuschprbnum").alias("mr_ltescpuschprbnum1"),functions.count("enodeb_id").alias("rows1"))
.selectExpr("ecell_id", "enodeb_id","scan_start_time1 as scan_start_time","insert_time1 as insert_time","mr_type1 as mr_type","mr_ltescphr1 as mr_ltescphr","mr_ltescpuschprbnum1 as mr_ltescpuschprbnum","rows1 as rows");

        Dataset<Row> ncRes = sparkSession.read().option("delimiter", "|").option("header", true).csv("/user/csv");Dataset<Row> mro=sparkSession.sql("。。。");Dataset<Row> ncJoinMro = ncRes.join(mro, mro.col("id").equalTo(ncRes.col("id")).and(mro.col("calid").equalTo(ncRes.col("calid"))), "left_outer").select(ncRes.col("id").as("int_id"), mro.col("vendor_id"),。。。                 );

leftDfWithWatermark.join(rightDfWithWatermark, 　　expr(""" leftDfId = rightDfId AND leftDfTime >= rightDfTime AND leftDfTime <= rightDfTime + interval 1 hour"""), 　　joinType = "leftOuter" )

package com.dx.testbroadcast;import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;import java.io.*;public class Test {public static void main(String[] args) {String personPath = "E:\\person.csv";String personOrderPath = "E:\\personOrder.csv";//writeToPersion(personPath);//writeToPersionOrder(personOrderPath);
SparkConf conf = new SparkConf();SparkSession sparkSession = SparkSession.builder().config(conf).appName("test-broadcast-app").master("local[*]").getOrCreate();Dataset<Row> person = sparkSession.read().option("header", "true").option("inferSchema", "true") //是否自动推到内容的类型.option("delimiter", ",").csv(personPath).as("person");person.printSchema();Dataset<Row> personOrder = sparkSession.read().option("header", "true").option("inferSchema", "true") //是否自动推到内容的类型.option("delimiter", ",").csv(personOrderPath).as("personOrder");personOrder.printSchema();// Default `inner`. Must be one of:`inner`, `cross`, `outer`, `full`, `full_outer`, `left`, `left_outer`,`right`, `right_outer`, `left_semi`, `left_anti`.Dataset<Row> resultDs = personOrder.join(functions.broadcast(person), personOrder.col("personid").equalTo(person.col("id")),"left");resultDs.explain();        resultDs.show(10);}private static void writeToPersion(String personPath) {BufferedWriter personWriter = null;try {personWriter = new BufferedWriter(new FileWriter(personPath));personWriter.write("id,name,age,address\r\n");for (int i = 0; i < 10000; i++) {personWriter.write("" + i + ",person-" + i + "," + i + ",address-address-address-address-address-address-address" + i + "\r\n");}} catch (Exception e) {e.printStackTrace();} finally {if (personWriter != null) {try {personWriter.close();} catch (IOException e) {e.printStackTrace();}}}}private static void writeToPersionOrder(String personOrderPath) {BufferedWriter personWriter = null;try {personWriter = new BufferedWriter(new FileWriter(personOrderPath));personWriter.write("personid,name,age,address\r\n");for (int i = 0; i < 1000000; i++) {personWriter.write("" + i + ",person-" + i + "," + i + ",address-address-address-address-address-address-address" + i + "\r\n");}} catch (Exception e) {e.printStackTrace();} finally {if (personWriter != null) {try {personWriter.close();} catch (IOException e) {e.printStackTrace();}}}}
}

== Physical Plan ==
*(2) BroadcastHashJoin [personid#28], [id#10], LeftOuter, BuildRight
:- *(2) FileScan csv [personid#28,name#29,age#30,address#31] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/E:/personOrder.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<personid:int,name:string,age:int,address:string>
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)))+- *(1) Project [id#10, name#11, age#12, address#13]+- *(1) Filter isnotnull(id#10)+- *(1) FileScan csv [id#10,name#11,age#12,address#13] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/E:/person.csv], PartitionFilters: [], PushedFilters: [IsNotNull(id)], ReadSchema: struct<id:int,name:string,age:int,address:string>+--------+--------+---+--------------------+---+--------+---+--------------------+
|personid|    name|age|             address| id|    name|age|             address|
+--------+--------+---+--------------------+---+--------+---+--------------------+
|       0|person-0|  0|address-address-a...|  0|person-0|  0|address-address-a...|
|       1|person-1|  1|address-address-a...|  1|person-1|  1|address-address-a...|
|       2|person-2|  2|address-address-a...|  2|person-2|  2|address-address-a...|
|       3|person-3|  3|address-address-a...|  3|person-3|  3|address-address-a...|
|       4|person-4|  4|address-address-a...|  4|person-4|  4|address-address-a...|
|       5|person-5|  5|address-address-a...|  5|person-5|  5|address-address-a...|
|       6|person-6|  6|address-address-a...|  6|person-6|  6|address-address-a...|
|       7|person-7|  7|address-address-a...|  7|person-7|  7|address-address-a...|
|       8|person-8|  8|address-address-a...|  8|person-8|  8|address-address-a...|
|       9|person-9|  9|address-address-a...|  9|person-9|  9|address-address-a...|
+--------+--------+---+--------------------+---+--------+---+--------------------+
only showing top 10 rows

        person.createOrReplaceTempView("temp_person");personOrder.createOrReplaceTempView("temp_person_order");Dataset<Row> sqlResult = sparkSession.sql(" SELECT /*+ BROADCAST (t11) */" +" t11.id,t11.name,t11.age,t11.address," +" t10.personid as person_id,t10.name as persion_order_name" +" FROM temp_person_order as t10 " +" inner join temp_person as t11" +" on t11.id = t10.personid ");sqlResult.show(10);sqlResult.explain();

+---+--------+---+--------------------+---------+------------------+
| id|    name|age|             address|person_id|persion_order_name|
+---+--------+---+--------------------+---------+------------------+
|  0|person-0|  0|address-address-a...|        0|          person-0|
|  1|person-1|  1|address-address-a...|        1|          person-1|
|  2|person-2|  2|address-address-a...|        2|          person-2|
|  3|person-3|  3|address-address-a...|        3|          person-3|
|  4|person-4|  4|address-address-a...|        4|          person-4|
|  5|person-5|  5|address-address-a...|        5|          person-5|
|  6|person-6|  6|address-address-a...|        6|          person-6|
|  7|person-7|  7|address-address-a...|        7|          person-7|
|  8|person-8|  8|address-address-a...|        8|          person-8|
|  9|person-9|  9|address-address-a...|        9|          person-9|
+---+--------+---+--------------------+---------+------------------+
only showing top 10 rows19/06/24 09:35:50 INFO FileSourceStrategy: Pruning directories with:
19/06/24 09:35:50 INFO FileSourceStrategy: Post-Scan Filters: isnotnull(personid#28)
19/06/24 09:35:50 INFO FileSourceStrategy: Output Data Schema: struct<personid: int, name: string>
19/06/24 09:35:50 INFO FileSourceScanExec: Pushed Filters: IsNotNull(personid)
19/06/24 09:35:50 INFO FileSourceStrategy: Pruning directories with:
19/06/24 09:35:50 INFO FileSourceStrategy: Post-Scan Filters: isnotnull(id#10)
19/06/24 09:35:50 INFO FileSourceStrategy: Output Data Schema: struct<id: int, name: string, age: int, address: string ... 2 more fields>
19/06/24 09:35:50 INFO FileSourceScanExec: Pushed Filters: IsNotNull(id)
== Physical Plan ==
*(2) Project [id#10, name#11, age#12, address#13, personid#28 AS person_id#94, name#29 AS persion_order_name#95]
+- *(2) BroadcastHashJoin [personid#28], [id#10], Inner, BuildRight:- *(2) Project [personid#28, name#29]:  +- *(2) Filter isnotnull(personid#28):     +- *(2) FileScan csv [personid#28,name#29] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/E:/personOrder.csv], PartitionFilters: [], PushedFilters: [IsNotNull(personid)], ReadSchema: struct<personid:int,name:string>+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)))+- *(1) Project [id#10, name#11, age#12, address#13]+- *(1) Filter isnotnull(id#10)+- *(1) FileScan csv [id#10,name#11,age#12,address#13] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/E:/person.csv], PartitionFilters: [], PushedFilters: [IsNotNull(id)], ReadSchema: struct<id:int,name:string,age:int,address:string>
19/06/24 09:35:50 INFO SparkContext: Invoking stop() from shutdown hook