Quy trình phát triển

1. Lấy một môi trường thực thi (execution environment)
2. Tải/ tạo dữ liệu ban đầu
3. Xác định các phép biến đổi dữ liệu
4. Chỉ định nơi lưu trữ kết quả tính toán
5. Kích hoạt thực thi chương trình

Ví dụ đếm từ (Word Count)

object TextFrequencyAnalyzer {
  def main(args: Array[String]): Unit = {
    // Khởi tạo môi trường thực thi
    val executionEnv = ExecutionEnvironment.getExecutionEnvironment
    
    // Tải dữ liệu văn bản
    val inputData = executionEnv.fromElements(
      "Có ai đó ở đó không?",
      "Tôi nghĩ tôi nghe thấy họ. Đứng lại! Có ai đó ở đó không?")
    
    // Biến đổi dữ liệu
    val tokenizedText = inputData.flatMap(sentence => sentence.toLowerCase().split("\\W+"))
    val filteredWords = tokenizedText.filter(word => word.nonEmpty)
    val wordPairs = filteredWords.map(word => (word, 1))
    val groupedWords = wordPairs.groupBy(0)
    val wordCounts = groupedWords.sum(1)
    
    // Lưu kết quả
    wordCounts.writeAsText(args(0))
    
    // Thực thi chương trình
    executionEnv.execute("Đếm tần suất từ")
  }
}

Đóng gói và triển khai lên YARN

Thêm plugin Maven để đóng gói

<build>
    <sourceDirectory>src/main/java</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
    <plugins>

        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>2.5.1</version>
            <configuration>
                <source>1.7</source>
                <target>1.7</target>
            </configuration>
        </plugin>

        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.0</version>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>testCompile</goal>
                    </goals>
                    <configuration>
                        <args>
                            <arg>-dependencyfile</arg>
                            <arg>${project.build.directory}/.scala_dependencies</arg>
                        </args>
                    </configuration>
                </execution>
            </executions>
        </plugin>
        
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>2.3</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <filters>
                            <filter>
                                <artifact>*:*</artifact>
                                <excludes>
                                    <exclude>META-INF/*.SF</exclude>
                                    <exclude>META-INF/*.DSA</exclude>
                                    <exclude>META-INF/*.RSA</exclude>
                                </excludes>
                            </filter>
                        </filters>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>com.example.TextAnalyzer</mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Triển khai lên YARN

Sử dụng lệnh rz để tải tệp JAR lên máy chủ, sau đó thực thi chương trình:

bin/flink run -m yarn-cluster -yn 2 /home/user/flinkjars/text_analyzer-1.0-SNAPSHOT.jar com.example.TextAnalyzer

Trang YARN cổng 8088 cho phép theo dõi trạng thái chương trình đã gửi. Kết quả đầu ra sẽ được lưu tại thư mục /opt/flink/flinkJAR.