Приложение слушает 2 темы кафки
userevent
paymentevent
Полезная нагрузка для использованияrevent
{"userId":"Id_223","firstname":"fname_223","lastname":"lname_223","phonenumber":"P98202384_223","usertimestamp":"Apr 5, 2019 2:58:47 PM"}
Полезная нагрузка для payventevent
{"paymentUserId":"Id_227","amount":1227.0,"location":"location_227","paymenttimestamp":"Apr 5, 2019 3:00:03 PM"}
На основе userId = paymentuserid, нам нужно объединить запись.
Кажется, приложение не может проанализировать запись из темы Кафки.
Должно быть что-то на from_json, которого мне не хватает.
Может ли кто-нибудь предоставить обратную связь?
Вот вывод консоли без какого-либо соединения. нет записей.
+------+---------+--------+-----------+-------------+-------------+------+--------+----------------+
|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);
}
}