У меня есть проект Flink, который получает потоки событий и выполняет некоторую логику для добавления флага этого события, а затем сохраняет flag и eventID на некоторое время для повторного использования или для запроса другим система. в...

Я использую API таблиц на Flink 1.4.0. У меня есть Table объекты, которые нужно преобразовать в DataSet типа Row . Проект был построен с использованием Maven и импортирован в IntelliJ. У меня есть следующий код, и среда IDE не может разрешить...

Я совершенно не знаком с Flink. Возможно, этот вопрос повторяется, но я нашел только одну ссылку, которая мне непонятна. https://stackoverflow.com/a/44294980/6904987 Я сохранил данные в Redis в примере формата ключевого значения. Ключ - это...

Я обрабатываю поток Kafka с помощью Flink SQL, где каждое сообщение извлекается из Kafka, обрабатывается с помощью flink sql и отправляется обратно в kafka. Мне нужен вложенный вывод, где вход плоский, а выход вложенный. Скажем, например, мой вклад...

Я загрузил файл Avro в набор данных Flink: AvroInputFormat<GenericRecord> test = new AvroInputFormat<GenericRecord>( new Path("PathToAvroFile") , GenericRecord.class); DataSet<GenericRecord> DS =...

Я слежу за Пример определения функции темпоральной таблицы , а компилятор отказывается принимать этот код: TemporalTableFunction rates = ratesHistory.createTemporalTableFunction("r_proctime", "r_currency"); tEnv.registerFunction("Rates",...

При использовании Flink Table SQL в моем проекте я обнаружил, что если в моем SQL есть какое-либо предложение GROUP BY , размер контрольной точки значительно увеличится. Например, INSERT INTO COMPANY_POST_DAY SELECT sta_date,...

Я пытаюсь использовать KeyedProcessFunction, но переменная ctx: Context в processFunction внутри моего KeyedProcessFunction возвращает значение null. Обратите внимание, что я использую значение по умолчанию TimeCharacteristic , равное...

Я пытаюсь проверить JSONObject с набором правил, если json соответствует набору правил, он вернет соответствующее правило, а JSONObject, если нет, он вернет JSONObject в Sideoutput, все это обрабатывается в ProcessFuntion, я получаю основной вывод,...

Я пытаюсь прочитать данные из темы kafka в DataStream и зарегистрировать DataStream , после чего использовать TableEnvironment.sqlQuery ("SQL") для запроса данные, когда TableEnvironment.execute () нет ошибки и нет вывода. public static...

Мое требование - создать триггер на основе двух событий (EVT_A и EVT_B независимо от порядка). Вот ожидание 1. EVT_A arrived. --> No action 2. EVT_B arrived --> Should Trigger 3. EVT_B arrived --> should Trigger since A was received...

У меня есть вариант использования, в котором я хочу соединить ключ с неключевым атрибутом в двух потоках данных в flink. Возможно ли это сделать?

У меня возникли проблемы с чтением журнала изменений Debezium из потока Kinesis. Могу ли я получить представление о том, как я могу анализировать события журнала изменений с помощью Flink SQL. Ниже представлена ​​моя попытка проанализировать поток...