Флинк. Данные пакетной загрузки. Исключение ООМ ⇐ JAVA

[Anonymous]

Я попробовал перенести данные из одной таблицы в другую во Flink с помощью SQL в пакетном режиме и заметил, что сначала все данные пытаются кэшироваться в памяти, прежде чем идти дальше по конвейеру. Поскольку у меня много данных, я получаю сообщение об ошибке нехватки памяти. Есть ли способ читать данные порциями, а не все сразу?

Пример потока:

CREATE TABLE t_destination_nrt ( `fk_id` decimal(32, 0) первичный ключ не применяется, строка `fv_text`, временная метка `fd_timestamp`, строка `fv_desc` ) С ( 'коннектор' = 'jdbc', 'url' = 'jdbc:postgresql://192.168.1.17:5432/postgres', 'имя-таблицы' = 'nrt.t_destination', 'имя пользователя' = 'postgres', 'пароль' = 'postgres', 'scan.fetch-size' = '1000', 'lookup.cache.ttl' = '60-е', 'lookup.cache.max-rows' = '10000', 'lookup.max-retries' = '3', 'sink.buffer-flush.max-rows' = '5000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' ) СОЗДАТЬ ТАБЛИЦУ t_source_nrt ( `fk_id` decimal(32, 0) первичный ключ не применяется, строка `fv_text`, временная метка `fd_timestamp` ) С ( 'коннектор' = 'jdbc', 'url' = 'jdbc:postgresql://192.168.1.17:5432/postgres', 'имя-таблицы' = 'nrt.t_source', 'имя пользователя' = 'postgres', 'пароль' = 'postgres', 'scan.fetch-size' = '1000', 'lookup.cache.ttl' = '60-е', 'lookup.cache.max-rows' = '10000', 'lookup.max-retries' = '3', 'sink.buffer-flush.max-rows' = '5000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' ) СОЗДАТЬ ТАБЛИЦУ t_dict_nrt ( `fk_id` decimal(32, 0) первичный ключ не применяется, строка `fv_desc` ) С ( 'коннектор' = 'jdbc', 'url' = 'jdbc:postgresql://192.168.1.17:5432/postgres', 'имя-таблицы' = 'nrt.t_dict', 'имя пользователя' = 'postgres', 'пароль' = 'postgres' ) вставить в t_destination_nrt (fk_id, fv_text, fd_timestamp, fv_desc) выберите /*+ ТРАНСЛЯЦИЯ(dn) */ sn.fk_id, sn.fv_text, sn.fd_timestamp, dn.fv_desc из t_source_nrt sn left join t_dict_nrt dn на sn.fk_id % 3 = dn.fk_id Я пытаюсь перезагрузить данные с помощью flink sql в пакетном режиме из таблицы в таблицу. Я ожидал, что данные будут загружаться частями, но flink выполняет полное кэширование исходной таблицы перед вставкой в ​​пункт назначения.