应用听力2Kafka主题
用户事件
付款事件
userevent的负载

{"userId":"Id_223","firstname":"fname_223","lastname":"lname_223","phonenumber":"P98202384_223","usertimestamp":"Apr 5, 2019 2:58:47 PM"}

paymentevent的有效负载

{"paymentUserId":"Id_227","amount":1227.0,"location":"location_227","paymenttimestamp":"Apr 5, 2019 3:00:03 PM"}

根据userid=paymentuserid，我们需要合并记录。
应用程序似乎无法解析Kafka主题中的记录。
我一定是漏掉了什么东西。
有人能提供早期反馈吗？
这里是没有任何连接发生的控制台输出。没有记录。

+------+---------+--------+-----------+-------------+-------------+------+--------+----------------+
|userId|firstname|lastname|phonenumber|usertimestamp|paymentuserId|amount|location|paymenttimestamp|
+------+---------+--------+-----------+-------------+-------------+------+--------+----------------+
+------+---------+--------+-----------+-------------+-------------+------+--------+----------------+

这是密码。

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 org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import static org.apache.spark.sql.functions.expr;

@SpringBootApplication
public class Stream2StreamJoin  implements CommandLineRunner{

    private static final Logger LOGGER =
              LoggerFactory.getLogger(Stream2StreamJoin.class);

    @Value("${kafka.bootstrap.server}")
    private String bootstrapServers;

    @Value("${kafka.userevent}")
    private String usereventTopic;

    @Value("${kafka.paymentevent}")
    private String paymenteventTopic;

    public void processData() {

        System.out.println(bootstrapServers);
        System.out.println(usereventTopic);
        System.out.println(paymenteventTopic);

        LOGGER.info(bootstrapServers);
        LOGGER.info(usereventTopic);
        LOGGER.info(paymenteventTopic);

        SparkConf sparkConf = new SparkConf().setAppName("Stream2StreamJoin").setMaster("local[*]");

        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(10));

        SparkSession spark = SparkSession
                  .builder()
                  .appName("Stream2StreamJoin")
                  .getOrCreate();

        spark.sparkContext().setLogLevel("ERROR");

        StructType userSchema =  DataTypes.createStructType(new StructField[] { 
                DataTypes.createStructField("userId", DataTypes.StringType, true),
                DataTypes.createStructField("firstname", DataTypes.StringType, true),
                DataTypes.createStructField("lastname", DataTypes.StringType, true),
                DataTypes.createStructField("phonenumber", DataTypes.StringType, true),
                DataTypes.createStructField("usertimestamp", DataTypes.TimestampType, true)
                });

        StructType paymentSchema =  DataTypes.createStructType(new StructField[] { 
                DataTypes.createStructField("paymentuserId", DataTypes.StringType, true),
                DataTypes.createStructField("amount", DataTypes.StringType, true),
                DataTypes.createStructField("location", DataTypes.StringType, true),                
                DataTypes.createStructField("paymenttimestamp", DataTypes.TimestampType, true)
                });

        Dataset<Row> userDataSet=spark.readStream().format("kafka")
                  .option("kafka.bootstrap.servers", bootstrapServers)
                  .option("subscribe", usereventTopic)
                  .option("startingOffsets", "earliest")
                  .load().selectExpr("CAST(value  AS STRING) as userEvent")
                     .select(functions.from_json(functions.col("userEvent"),userSchema).as("user"))
                     .select("user.*")
                     ; 

        Dataset<Row> paymentDataSet=spark.readStream().format("kafka")
                  .option("kafka.bootstrap.servers", bootstrapServers)
                  .option("subscribe", paymenteventTopic)
                  .option("startingOffsets", "earliest")
                  .load().selectExpr("CAST( value AS STRING) as paymentEvent")
                     .select(functions.from_json(functions.col("paymentEvent"),paymentSchema).as("payment"))
                     .select("payment.*")
                     ;

        Dataset<Row> userDataSetWithWatermark = userDataSet.withWatermark("usertimestamp", "2 hours");

        Dataset<Row> paymentDataSetWithWatermark = paymentDataSet.withWatermark("paymenttimestamp", "3 hours");

        Dataset<Row> joindataSet =  userDataSetWithWatermark.join(
                paymentDataSetWithWatermark,
                  expr(
                          "userId = paymentuserId AND usertimestamp >= paymenttimestamp AND usertimestamp <= paymenttimestamp + interval 1 hour")
                );

        joindataSet.writeStream().format("console").start();

        try {

            spark.streams().awaitAnyTermination();
        } catch (StreamingQueryException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        }

    @Override
    public void run(String... args) throws Exception {
        processData();

    }

    public static void main(String[] args) throws Exception {

        System.setProperty("hadoop.home.dir", "/Users/workspace/java/spark-kafka-streaming");

        SpringApplication.run(Stream2StreamJoin.class, args);
    }

}