Начало работы с Flink (3) — среда и развертывание

file

Flink – это платформа для потоковой передачи больших данных с открытым исходным кодом. Она может одновременно выполнять пакетную и потоковую передачу. Она обладает такими преимуществами, как отказоустойчивость, высокая пропускная способность и низкая задержка. В этой статье кратко описаны этапы установки flink в Windows и Linux, а также примеры программ.Выполнение, включая локальную среду отладки, кластерную среду. Кроме того, вводится строительство девелоперского проекта Флинка.

Прежде всего, чтобы запустить Flink, нам нужно скачать и распаковать бинарный пакет Flink, адрес загрузки выглядит следующим образом: https://flink.apache.org/downloads.html.

Мы можем выбрать комбинированную версию Flink и Scala, здесь мы выбираем последнюю версию 1.9.Apache Flink 1.9.0 for Scala 2.12Скачать.

После успешной загрузки вы можете запустить Flink в системе Windows через bat-файл Windows или Cygwin.

В системе Linux она делится на одну машину, кластер, Hadoop и другие ситуации.

Запуск через bat-файл Windows

Сначала запустите окно командной строки cmd, войдите в папку flink и запустите каталог bin.start-cluster.bat

Примечание. Для запуска flink требуется среда java. Убедитесь, что в системе настроена переменная среды java.

$ cd flink
$ cd bin
$ start-cluster.bat
Starting a local cluster with one JobManager process and one TaskManager process.
You can terminate the processes via CTRL-C in the spawned shell windows.
Web interface by default on http://localhost:8081/.

file

После того, как будет показано, что запуск прошел успешно, мы можем увидеть страницу управления flink, посетив http://localhost:8081/ в браузере.

Запуск через Cygwin

CygwinЭто UNIX-подобная среда моделирования, работающая на платформе Windows.Загрузить с официального сайта: http://cygwin.com/install.html

После успешной установки запустите терминал Cygwin и запуститеstart-cluster.shсценарий.

$ cd flink
$ bin/start-cluster.sh
Starting cluster.

file

Установите flink в системе Linux

Установка одного узла

Установка одного узла в Linux такая же, как и в cygwin, загрузитеApache Flink 1.9.0 for Scala 2.12, а затем просто запустите start-cluster.sh после распаковки.

Установка кластера

Установка кластера делится на следующие этапы:

1. Скопируйте распакованный каталог flink на каждую машину.

2. Выберите один в качестве главного узла, а затем измените все машины conf/flink-conf.yaml

jobmanager.rpc.address = master主机名

3. Измените conf/slaves, чтобы записать все рабочие узлы

work01
work02

4. Запускаем кластер на мастере

bin/start-cluster.sh

Установить на Хадуп

Мы можем выбрать запуск Flink в кластере Yarn.

Загрузите пакет Flink для Hadoop

Убедитесь, что HADOOP_HOME установлен правильно.

Запустите bin/yarn-session.sh

Запустите пример программы flink

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

Отправьте пример пакетной программы flink:

bin/flink run examples/batch/WordCount.jar

Это пакетный образец программы по примерам, предоставленным flink, который подсчитывает количество слов.

$ bin/flink run examples/batch/WordCount.jar
Starting execution of program
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
(a,5)
(action,1)
(after,1)
(against,1)
(all,2)
(and,12)
(arms,1)
(arrows,1)
(awry,1)
(ay,1)

Результат получен, здесь считается набор данных по умолчанию, а ввод и вывод можно указать с помощью --input --output.

Мы можем видеть, что происходит на странице:

file

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

Запустите NC-сервер:

nc -l 9000

Отправьте пример пакетной программы flink:

bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000

Это пример программы потоковой обработки по примерам, предоставленным flink, которая получает входящие данные сокета и подсчитывает количество слов.

написать слово на нк стороне

$ nc -l 9000
lorem ipsum
ipsum ipsum ipsum
bye

вывод в лог

$ tail -f log/flink-*-taskexecutor-*.out
lorem : 1
bye : 1
ipsum : 4

остановить мерцание

$ ./bin/stop-cluster.sh

После установки Flink вы можете легко запустить Flink, если быстро создадите проект Flink и завершите разработку соответствующего кода.

инструменты для сборки

Проекты Flink можно создавать с использованием различных инструментов сборки. Чтобы быстро приступить к работе, Flink предоставляет шаблоны проектов для следующих инструментов сборки:

Maven
Gradle

Эти шаблоны помогают структурировать проект и создавать начальные файлы сборки.

Maven

Требования к окружающей среде

Единственное требование - использоватьMaven 3.0.4(или позже) и установитьJava 8.x.

Создать проект

Используйте одну из следующих команд, чтобыСоздать проект:

Использование архетипов Maven

 $ mvn archetype:generate                               \
      -DarchetypeGroupId=org.apache.flink              \
      -DarchetypeArtifactId=flink-quickstart-java      \
      -DarchetypeVersion=1.9.0

Запустите скрипт быстрого запуска

 curl https://flink.apache.org/q/quickstart.sh | bash -s 1.9.0

После завершения загрузки просмотрите структуру каталогов проекта:

tree quickstart/
quickstart/
├── pom.xml
└── src
    └── main
        ├── java
        │   └── org
        │       └── myorg
        │           └── quickstart
        │               ├── BatchJob.java
        │               └── StreamingJob.java
        └── resources
            └── log4j.properties

Примерный проект представляет собойMaven project, который содержит два класса:StreamingJobиBatchJobсоответственноDataStream and DataSetБазовый скелет программы.mainМетоды — это точки входа в программы как для тестирования/выполнения IDE, так и для развертывания.

Мы рекомендуем вамИмпортируйте этот проект в IDEразработать и протестировать его. IntelliJ IDEA поддерживает проекты Maven из коробки. Если вы используете Eclipse, используйтеm2e-плагинМожетИмпортировать проект Maven. Некоторые пакеты Eclipse включают этот подключаемый модуль по умолчанию, другие требуют, чтобы вы установили его вручную.

Пожалуйста, обрати внимание: Память кучи JVM по умолчанию может быть слишком мала для Flink, вам следует увеличить объем кучи вручную. В затмении выберитеRun Configurations -> ArgumentsИ вVM ArgumentsНапишите в соответствующем поле ввода:-Xmx800m. В IntelliJ IDEA рекомендуется выбрать из менюHelp | Edit Custom VM Optionsдля изменения параметров JVM.

Построить проект

Если ты хочешьсобрать/упаковать свой проект, пожалуйста, запустите ' в каталоге проектаmvn clean package' Заказ. После выполнения команды выНайдите JAR-файл, который содержит ваше приложение, а также соединители и библиотеки, которые были добавлены к приложению в качестве зависимостей:target/-.jar.

Уведомление:Если вы используете другие классы вместоStreamingJobВ качестве основного класса/записи вашего приложения мы рекомендуем соответствующим образом изменитьpom.xmlв файлеmainClassконфигурация. Таким образом, Flink может запускать приложение из JAR-файла без дополнительного указания основного класса.

Gradle

Требования к окружающей среде

Единственное требование - использоватьGradle 3.x(или позже) и установитьJava 8.x.

Создать проект

Используйте одну из следующих команд, чтобыСоздать проект:

Пример Gradle:

build.gradle

buildscript {
    repositories {
        jcenter() // this applies only to the Gradle 'Shadow' plugin
    }
    dependencies {
        classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.4'
    }
}

plugins {
    id 'java'
    id 'application'
    // shadow plugin to produce fat JARs
    id 'com.github.johnrengelman.shadow' version '2.0.4'
}


// artifact properties
group = 'org.myorg.quickstart'
version = '0.1-SNAPSHOT'
mainClassName = 'org.myorg.quickstart.StreamingJob'
description = """Flink Quickstart Job"""

ext {
    javaVersion = '1.8'
    flinkVersion = '1.9.0'
    scalaBinaryVersion = '2.11'
    slf4jVersion = '1.7.7'
    log4jVersion = '1.2.17'
}


sourceCompatibility = javaVersion
targetCompatibility = javaVersion
tasks.withType(JavaCompile) {
    options.encoding = 'UTF-8'
}

applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"]

task wrapper(type: Wrapper) {
    gradleVersion = '3.1'
}

// declare where to find the dependencies of your project
repositories {
    mavenCentral()
    maven { url "https://repository.apache.org/content/repositories/snapshots/" }
}

// 注意：我们不能使用 "compileOnly" 或者 "shadow" 配置，这会使我们无法在 IDE 中或通过使用 "gradle run" 命令运行代码。
// 我们也不能从 shadowJar 中排除传递依赖（请查看 https://github.com/johnrengelman/shadow/issues/159)。
// -> 显式定义我们想要包含在 "flinkShadowJar" 配置中的类库!
configurations {
    flinkShadowJar // dependencies which go into the shadowJar

    // 总是排除这些依赖（也来自传递依赖），因为 Flink 会提供这些依赖。
    flinkShadowJar.exclude group: 'org.apache.flink', module: 'force-shading'
    flinkShadowJar.exclude group: 'com.google.code.findbugs', module: 'jsr305'
    flinkShadowJar.exclude group: 'org.slf4j'
    flinkShadowJar.exclude group: 'log4j'
}

// declare the dependencies for your production and test code
dependencies {
    // --------------------------------------------------------------
    // 编译时依赖不应该包含在 shadow jar 中，
    // 这些依赖会在 Flink 的 lib 目录中提供。
    // --------------------------------------------------------------
    compile "org.apache.flink:flink-java:${flinkVersion}"
    compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"

    // --------------------------------------------------------------
    // 应该包含在 shadow jar 中的依赖，例如：连接器。
    // 它们必须在 flinkShadowJar 的配置中！
    // --------------------------------------------------------------
    //flinkShadowJar "org.apache.flink:flink-connector-kafka-0.11_${scalaBinaryVersion}:${flinkVersion}"

    compile "log4j:log4j:${log4jVersion}"
    compile "org.slf4j:slf4j-log4j12:${slf4jVersion}"

    // Add test dependencies here.
    // testCompile "junit:junit:4.12"
}

// make compileOnly dependencies available for tests:
sourceSets {
    main.compileClasspath += configurations.flinkShadowJar
    main.runtimeClasspath += configurations.flinkShadowJar

    test.compileClasspath += configurations.flinkShadowJar
    test.runtimeClasspath += configurations.flinkShadowJar

    javadoc.classpath += configurations.flinkShadowJar
}

run.classpath = sourceSets.main.runtimeClasspath

jar {
    manifest {
        attributes 'Built-By': System.getProperty('user.name'),
                'Build-Jdk': System.getProperty('java.version')
    }
}

shadowJar {
    configurations = [project.configurations.flinkShadowJar]
}

setting.gradle

rootProject.name = 'quickstart'

или запустите скрипт быстрого запуска

    bash -c "$(curl https://flink.apache.org/q/gradle-quickstart.sh)" -- 1.9.0 2.11

Ознакомьтесь со структурой каталогов:

tree quickstart/
quickstart/
├── README
├── build.gradle
├── settings.gradle
└── src
    └── main
        ├── java
        │   └── org
        │       └── myorg
        │           └── quickstart
        │               ├── BatchJob.java
        │               └── StreamingJob.java
        └── resources
            └── log4j.properties

Примерный проект представляет собойGradle-проект, который содержит два класса:StreamingJobиBatchJobдаDataStreamиDataSetБазовый скелет программы.mainМетод — это точка входа в программу либо для тестирования/выполнения IDE, либо для развертывания.

Мы рекомендуем вамИмпортируйте этот проект в свою IDEразработать и протестировать его. IntelliJ IDEA устанавливаетсяGradleПост-плагинная поддержка проектов Gradle. Затмение черезEclipse BuildshipПлагин поддерживает проекты Gradle (с учетомshadowУ плагина есть требования к версии Gradle, обязательно укажите версию Gradle >= 3.0 на последнем шаге мастера импорта). вы также можете использоватьИнтеграция Gradle с IDEСоздайте файлы проекта из Gradle.

Построить проект

Если ты хочешьсобрать/упаковать проект, пожалуйста, запустите ' в каталоге проектаgradle clean shadowJar' Заказ. После выполнения команды выНайдите JAR-файл, который содержит ваше приложение, а также соединители и библиотеки, которые были добавлены к приложению в качестве зависимостей:build/libs/--all.jar.

Уведомление:Если вы используете другие классы вместоStreamingJobВ качестве основного класса/записи вашего приложения мы рекомендуем соответствующим образом изменитьbuild.gradleв файлеmainClassNameконфигурация. Таким образом, Flink может запускать приложение из JAR-файла без дополнительного указания основного класса.

Серия статей Flink:

Введение в Flink (1) — Введение в Apache Flink Введение в Flink (2) — Введение в архитектуру Flink

Больше вычислений в реальном времени, Flink, Kafka и других связанных технических сообщений в блогах, добро пожаловать, чтобы обратить внимание на потоковые вычисления в реальном времени.

file