ApacheSparkを使用したSpringCloudデータフロー

投稿日: 2019-10-19 2022-10-31
タグ: Java, Spark

1. 序章

Spring Cloud Data Flowは、データ統合とリアルタイムデータ処理パイプラインを構築するためのツールキットです。

この場合のパイプラインは、 Spring CloudStreamまたはSpringCloudTaskフレームワークを使用して構築されたSpringBootアプリケーションです。

このチュートリアルでは、Spring CloudDataFlowをApacheSparkで使用する方法を示します。

2. データフローローカルサーバー

まず、ジョブをデプロイできるようにするには、データフローサーバーを実行する必要があります。

データフローサーバーをローカルで実行するには、spring-cloud-starter-dataflow-server-local依存関係を使用して新しいプロジェクトを作成する必要があります。

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-dataflow-server-local</artifactId>
    <version>1.7.4.RELEASE</version>
</dependency>

その後、サーバーのメインクラスに@EnableDataFlowServerでアノテーションを付ける必要があります。

@EnableDataFlowServer
@SpringBootApplication
public class SpringDataFlowServerApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(
          SpringDataFlowServerApplication.class, args);
    }
}

このアプリケーションを実行すると、ポート9393にローカルデータフローサーバーが作成されます。

3. プロジェクトの作成

スタンドアロンのローカルアプリケーションとしてSparkJob を作成し、クラスターを実行する必要がないようにします。

3.1. 依存関係

まず、Spark依存関係を追加します。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.10</artifactId>
    <version>2.4.0</version>
</dependency>

3.2. ジョブの作成

そして私たちの仕事のために、円周率を概算しましょう：

public class PiApproximation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("BaeldungPIApproximation");
        JavaSparkContext context = new JavaSparkContext(conf);
        int slices = args.length >= 1 ? Integer.valueOf(args[0]) : 2;
        int n = (100000L * slices) > Integer.MAX_VALUE ? Integer.MAX_VALUE : 100000 * slices;

        List<Integer> xs = IntStream.rangeClosed(0, n)
          .mapToObj(element -> Integer.valueOf(element))
          .collect(Collectors.toList());

        JavaRDD<Integer> dataSet = context.parallelize(xs, slices);

        JavaRDD<Integer> pointsInsideTheCircle = dataSet.map(integer -> {
           double x = Math.random() * 2 - 1;
           double y = Math.random() * 2 - 1;
           return (x * x + y * y ) < 1 ? 1: 0;
        });

        int count = pointsInsideTheCircle.reduce((integer, integer2) -> integer + integer2);

        System.out.println("The pi was estimated as:" + count / n);

        context.stop();
    }
}

4. データフローシェル

データフローシェルは、サーバーとの対話を可能にするアプリケーションです。シェルはDSLコマンドを使用してデータフローを記述します。

データフローシェルを使用するには、それを実行できるプロジェクトを作成する必要があります。まず、spring-cloud-dataflow-shell依存関係が必要です。

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-dataflow-shell</artifactId>
    <version>1.7.4.RELEASE</version>
</dependency>

依存関係を追加した後、データフローシェルを実行するクラスを作成できます。

@EnableDataFlowShell
@SpringBootApplication
public class SpringDataFlowShellApplication {
     
    public static void main(String[] args) {
        SpringApplication.run(SpringDataFlowShellApplication.class, args);
    }
}

5. プロジェクトの展開

プロジェクトをデプロイするために、ApacheSparkでcluster 、 yarn 、およびclientの3つのバージョンで使用できるいわゆるタスクランナーを使用します。ローカルのクライアントバージョンを使用します。

タスクランナーは、Sparkジョブを実行するものです。

そのためには、まずデータフローシェルを使用してタスクを登録する必要があります。

app register --type task --name spark-client --uri maven://org.springframework.cloud.task.app:spark-client-task:1.0.0.BUILD-SNAPSHOT

このタスクでは、複数の異なるパラメーターを指定できます。これらのパラメーターの一部はオプションですが、Sparkジョブを適切にデプロイするために必要なパラメーターの一部は次のとおりです。

spark.app-class 、提出されたジョブのメインクラス
spark.app-jar 、私たちの仕事を含む太ったjarへのパス
spark.app- name 、私たちの仕事に使用される名前
spark.app-args 、ジョブに渡される引数

登録済みのタスクspark-clientを使用して、必要なパラメーターを指定することを忘れずに、ジョブを送信できます。

task create spark1 --definition "spark-client \
  --spark.app-name=my-test-pi --spark.app-class=com.baeldung.spring.cloud.PiApproximation \
  --spark.app-jar=/apache-spark-job-0.0.1-SNAPSHOT.jar --spark.app-args=10"

spark.app-jar は、私たちの仕事でのfat-jarへのパスであることに注意してください。

タスクが正常に作成されたら、次のコマンドを使用してタスクの実行に進むことができます。

task launch spark1

これにより、タスクの実行が呼び出されます。

6. 概要

このチュートリアルでは、Spring CloudDataFlowフレームワークを使用してApacheSparkでデータを処理する方法を示しました。 Spring Cloud Data Flowフレームワークの詳細については、ドキュメントを参照してください。

すべてのコードサンプルはGitHubにあります。

getdocs

13036