Spark Java Structured Streaming filters

Question

I have the following data below:

+-------+----------+------------+---------+---------------------+-----------+
|id     |resource id|resource name|event-desc|event-date       |ipaddress1  |
+-------+----------+------------+---------+---------------------+-----------+
|2010001|119       |Netopia     |logout   |+56975-05-07 23:01:37|25:34:21:44|
|2010001|119       |Netopia     |login    |+56975-05-07 23:01:37|25:34:21:44|
|2010001|119       |Netopia     |logout   |+56975-05-07 23:01:37|25:34:21:45|
|2010001|119       |Netopia     |login    |+56975-05-07 23:01:37|25:34:21:45|
|2010001|119       |Netopia     |logout   |+56975-05-07 23:01:37|25:34:21:44|
|2010001|119       |Netopia     |logout   |+56975-05-07 23:01:37|25:34:21:46|
+-------+----------+------------+---------+---------------------+-----------+

I need to apply multiple filters to same dataset and pass to Kafka.

Dataset<Row> mainData=df.select( "data.*").filter("data.eventdesc='logout'");

Send the filtered data to kafka, again there will be another filter condition.

mainData=df.select( "data.*").filter("data.eventdesc='login'");

send the data to kafka.

In this case only the last expression is working. Any suggestions on how I can filter data for both of the conditions and send to kafka?

Frank · Accepted Answer · 2025-03-18 13:09:14Z

0

Based on this answer.

Dataset<Row> mainData=df.select( "data.*").filter("data.eventdesc='logout' or data.eventdesc='login'")

answered Mar 18 at 13:09

Frank

6346 silver badges16 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

tzuwei93 · Accepted Answer · 2025-03-18 16:24:42Z

the variable (mainData) is reassigned and thus reference to a new object, I'd like to recommend below approach, if you have complex filter logic.

Dataset<Row> data = df.select("data.*")
    .filter(functions.col("data.eventdesc").equalTo("logout")
        .or(functions.col("data.eventdesc").equalTo("login")));

and, for passing a dataset to Kafka topics, you would need to import the related jar with proper version selected first, as below ( not for streaming data ). Hope it is helpful for you.

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    <version>3.5.x</version>
</dependency>

data.selectExpr("value").write()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("topic", "login-logout-events")
        .save();

Collectives™ on Stack Overflow

Spark Java Structured Streaming filters

2 Answers 2

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related