Are `Partition By` in the Table API and `KeyBy` in the DataS Apache Flink #troubleshooting

Are `Partition By` in the Table API and `KeyBy` in...

Maryam

01/24/2023, 7:48 PM

Are

Partition By

in the Table API and

KeyBy

in the DataStream API the same in terms of how they process data in streaming mode, specifically in terms of whether data is processed together in the same task slot
? Additionally, I am asking because my Top-N query seems to be producing non-deterministic results when using parallelism greater than 1.

sap1ens

01/24/2023, 7:55 PM

I know for sure they’re not the same for connector tables,

Partition By

doesn’t cause a shuffle.

👋 1

sap1ens

01/24/2023, 8:00 PM

table.exec.sink.keyed-shuffle

can help with a problem at the sink level.

Maryam

01/24/2023, 9:20 PM

I am using Datastream source & sink and use the table api for the Top-N query. The query result changes every time it executed as if data with same key don’t get processed together.

sap1ens

01/24/2023, 9:28 PM

Have you tried calling a

keyBy

right after the source?

Maryam

01/24/2023, 10:01 PM

you mean sth like this?

Copy code

val table = tableEnv.fromDataStream(dataStreamSource.keyBy(id))
tableEnv.createTemporaryView("table", table)

I was hoping

Partition By

in SQL would do that by default. 🤔

Copy code

val dedupeSqlQuery =
      """
        |SELECT *
        |FROM (
        |  SELECT *,
        |    ROW_NUMBER() OVER (PARTITION BY id ORDER BY processedCount DESC) AS row_num
        |  FROM table)
        |WHERE row_num = 1
        |""".stripMargin

sap1ens

01/24/2023, 10:03 PM

Yeah, something like that. I don’t know how

PARTITION BY

is implemented for window functions. I thought you’re referring to a clause when specifying a connector table, e.g. for Kafka or object store like S3, I know for sure that doesn’t cause a shuffle.

👍 1

Open in Slack

Previous Next