Файлы застряли как ⇐ JAVA

[Anonymous]

Я использую задание по потоковой передаче потока, используя Table API, который читается от Kafka и записывает S3 (сейчас я использую локальный путь для имитации S3). В задании используется разъем файловой системы для записи данных в формате Parquet. < /P>
Вот упрощенная версия моего кода: < /p>

Код: Выделить всё

tableEnv.executeSql(
"CREATE TABLE logs_source (" +
" `timestamp` STRING," +
" user_id INT," +
" message STRING" +
") WITH (" +
" 'connector' = 'kafka'," +
" 'topic' = 'kafka-5s'," +
" 'properties.bootstrap.servers' = 'localhost:9092'," +
" 'properties.group.id' = 'kafka-5s-1'," +
" 'scan.startup.mode' = 'earliest-offset'," +
" 'format' = 'json'," +
" 'json.timestamp-format.standard' = 'ISO-8601'" +
")"
);

tableEnv.executeSql(
"CREATE TABLE s3_sink (" +
" `timestamp` STRING," +
" user_id INT," +
" message STRING" +
") WITH (" +
" 'connector' = 'filesystem'," +
" 'path' = 'file:///D://flink-data/data'," +
" 'format' = 'parquet'," +
" 'parquet.compression' = 'SNAPPY'," +
" 'sink.rolling-policy.rollover-interval' = '1000'," +
" 'sink.rolling-policy.file-size' = '10'," +
" 'sink.rolling-policy.inactivity-interval' = '1000'" +
")"
);

tableEnv.executeSql("INSERT INTO s3_sink SELECT `timestamp`, user_id, message FROM logs_source");

Проблема:
Flink действительно пишут .Inprogress файлы в целевом каталоге (например, часть 0-0.InProgress), но эти файлы никогда не завершаются в правильные файлы.
. /> Настройка коротких переворот-интервалов, размера файлов и неактивности Interval (1000 мс и 10 байтов) для запуска быстрого броска. < /p>
< /li>
включенная контрольная точка с: < /p>
< /li>
< /ul>

Код: Выделить всё

        Configuration config = GlobalConfiguration.loadConfiguration(
"D:\\Gitlab\\java-project\\src\\main\\resources\\"
);

config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///D://flink-data/checkpoints/");

// Print out S3 credentials for verification
System.out.println("fs.s3a.access.key: " + config.getString("fs.s3a.access.key", "NOT SET"));
System.out.println("fs.s3a.secret.key: " + config.getString("fs.s3a.secret.key", "NOT SET"));
System.out.println("fs.s3a.endpoint: " + config.getString("fs.s3a.endpoint", "NOT SET"));

// Initialize the file system with the config
FileSystem.initialize(config, null);

// Create a streaming execution environment with checkpoints enabled
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
env.enableCheckpointing(10000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.configure(config);

// Create a TableEnvironment for streaming queries
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

проверял, что тема kafka (kafka-5s) непрерывно создает данные.>

Подробнее здесь: https://stackoverflow.com/questions/796 ... quet-files