Flink: Left joining a stream with a static list

Question

I want to join a streaming of Attempts to a static list of blocked emails and group the result by IP, so I can later count a pack of relevant stats. The result should be delivered as a sliding window of 30 minutes after each 10 seconds. Below is one of several ways I have tried to achieve this:

override fun performQuery(): Table {
    val query = "SELECT ip, " +
        "COUNT(CASE WHEN success IS false THEN 1 END) AS fails, " +
        "COUNT(CASE WHEN success IS true THEN 1 END) AS successes, " +
        "COUNT(DISTINCT id) accounts, " +
        "COUNT(CASE WHEN id = 0 THEN 1 END) AS non_existing_accounts, " +
        "COUNT(CASE WHEN blockedEmail IS NOT NULL THEN 1 END) AS blocked_accounts " +
        "FROM Attempts " +
        "LEFT JOIN LATERAL TABLE(blockedEmailsList()) AS T(blockedEmail) ON TRUE " +
        "WHERE Attempts.email <> '' AND Attempts.createdAt < CURRENT_TIMESTAMP " +
        "GROUP BY HOP(Attempts.createdAt, INTERVAL '10' SECOND, INTERVAL '30' MINUTE), ip"

    return runQuery(query)
        .select("ip, accounts, fails, successes, non_existing_accounts, blocked_accounts")
}

This uses the User-Defined Table Function below, which is already registered in my tableEnv as blockedEmailsList:

public class BlockedEmailsList extends TableFunction<Row> {
    private Collection<String> emails;

    public BlockedEmailsList(Collection<String> emails) {
        this.emails = emails;
    }

    public Row read(String email) {
        return Row.of(email);
    }

    public void eval() {
        this.emails.forEach(email -> collect(read(email)));
    }
}

However, it returns the error below:

Caused by: org.apache.flink.table.api.TableException: Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.

If I do as it suggests and cast the created_at to TIMESTAMP, I get this instead:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Window can only be defined over a time attribute column.

I have found other questions here on Stack Overflow related to theses exceptions, but they involve streams and temporal tables and none of them solves a case of joining a stream to a static list.

Any ideas?

EDIT: Looks like there is an open issue in the Flink project for my use case: https://cwiki.apache.org/confluence/display/FLINK/FLIP-17+Side+Inputs+for+DataStream+API

So, I'm also accepting workaround suggestions.

Answer 1

Caused by: org.apache.flink.table.api.TableException: Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.

The reason is lateral table function is a Flink regular join, and a regular join will send null value, for example

left:（K0, A）, right（K1, T1）  => send    (K0, A, NULL, NULL)
left:         , right（K0, T2） => retract (K0, A, NULL, NULL )  
                                   send   (K0, A, K0, T2)

and thus the time attribute from input stream will loss after join.

In your case, you do not need a TableFunction, you can use a Scalar Function like:

 public static class BlockedEmailFunction extends ScalarFunction {
     private static List<String> blockedEmails = ...;
     public Boolean eval(String email) {
        return blockedEmails.contains(attempt.getEmail());
     }
 }


// register function
env.createTemporarySystemFunction("blockedEmailFunction", BlockedEmailFunction.class);

// call registered function in SQL and do window operation as your expected
env.sqlQuery("SELECT blockedEmailFunction(email) as status, ip, createdAt FROM Attempts");
 

Answer 2

I managed to implement a workaround which solved my problem!

Instead of joining streamed Attempts with the static list of emails, I mapped beforehand each Attempt to a new one with an added blockedEmail attribute. If the static list blockedEmails contains the current Attempt email, I set its blockedEmail attribute to true.

DataStream<Attempt> attemptsStream = sourceApi.<Attempt>startStream().map(new MapFunction<Attempt, Attempt>() {
    @Override
    public Attempt map(Attempt attempt) throws Exception {
        if (blockedEmails.contains(attempt.getEmail())) {
            attempt.setBlockedEmail(true);
        }
        return attempt;
    }
});

The static list blockedEmails is of type HashSet, so a lookup would be O(1).

Finally, the grouping query was adjusted to:

override fun performQuery(): Table {
    val query = "SELECT ip, " +
        "COUNT(CASE WHEN success IS false THEN 1 END) AS fails, " +
        "COUNT(CASE WHEN success IS true THEN 1 END) AS successes, " +
        "COUNT(DISTINCT id) accounts, " +
        "COUNT(CASE WHEN id = 0 THEN 1 END) AS non_existing_accounts, " +
        "COUNT(CASE WHEN blockedEmail IS true THEN 1 END) AS blocked_accounts " +
        "FROM Attempts " +
        "WHERE Attempts.email <> '' " +
        "GROUP BY HOP(Attempts.createdAt, INTERVAL '10' SECOND, INTERVAL '30' MINUTE), ip"

    return runQuery(query)
        .select("ip, accounts, fails, successes, non_existing_accounts, blocked_accounts")
}

So far, the joining problem between streams and static lists seems unresolved yet, but in my case the above workaround solution solved it fine.