Apache Flink — это мощная платформа обработки потоков, которая обеспечивает обработку данных в реальном времени. Docker предоставляет простой способ локально настроить Apache Flink и поэкспериментировать с ним. В этой статье мы расскажем вам, как запустить Apache Flink с Docker, продемонстрируем, как интегрировать Apache Kafka с Flink с помощью Dockerfile, и предоставим пример сценария Flink с использованием Python для потоковой обработки.

Настройка Apache Flink с помощью Docker

Шаг 1:Установите Docker
Если Docker не установлен в вашей системе, вы можете следовать инструкциям в [официальной документации](https://docs.docker.com/get- docker/), чтобы установить его.

Шаг 2. Запустите контейнер Apache Flink
Выполните следующую команду в своем терминале, чтобы запустить контейнер Apache Flink:

docker run -d -p 8081:8081 apache/flink:1.14.0

Это извлечет образ Apache Flink и запустит контейнер с веб-панелью Flink, доступной по адресу `http://localhost:8081`.

Dockerfile для интеграции Apache Kafka и Flink

Шаг 1. Создайте Dockerfile
Создайте `Dockerfile` в выбранной вами папке со следующим содержимым:

FROM flink:1.14.0
# Install Kafka connector dependencies
RUN mkdir -p /opt/flink/usrlib
RUN wget -P /opt/flink/usrlib https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-kafka_2.12/1.14.0/flink-connector-kafka_2.12-1.14.0.jar

Шаг 2. Создайте и запустите образ Docker
Перейдите в каталог, содержащий `Dockerfile`, и выполните следующие команды:

docker build -t flink-kafka-integration .
docker run -it flink-kafka-integration

Это создаст образ Docker и запустит контейнер с интегрированным Apache Flink и соединителем Kafka.

Пример сценария Flink с использованием Python

Шаг 1. Установите Flink
Установите Python SDK для Apache Flink с помощью следующей команды:

pip install apache-flink

Шаг 2. Скрипт Python Flink
Вот пример скрипта Python, который демонстрирует, как использовать SDK Python Flink для потоковой обработки:

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.table.descriptors import Schema, Kafka
# Create Flink environment
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)
# Define Kafka source
t_env.connect(
    Kafka()
    .version("universal")
    .topic("input_topic")
    .start_from_latest()
    .property("bootstrap.servers", "localhost:9092")
).with_format(
    Json()
).with_schema(
    Schema().field("field1", "STRING").field("field2", "INT")
).create_temporary_table("kafka_source")
# Define processing logic
result = t_env.from_path("kafka_source").select("field1, field2 + 1")
# Write the result to a Kafka sink
result.execute_insert("output_topic")
# Execute Flink job
env.execute("Python Flink Job")

Запуск Apache Flink с Docker предоставляет эффективный способ экспериментировать с потоковой обработкой. Интеграция Apache Kafka с Flink расширяет возможности обработки данных в реальном времени. Использование сценариев Python с Python SDK от Flink позволяет легко создавать и развертывать задания потоковой обработки. Следуя этому руководству и используя предоставленные примеры сценариев, вы сможете изучить возможности Apache Flink для потоковой обработки и анализа данных в своих проектах.