E-MapReduce:Lire et écrire dans Kafka

Étape 1 : Télécharger les fichiers JAR Kafka vers OSS

Les versions du moteur EMR Serverless Spark esr-2.8.0, esr-3.4.0, esr-4.4.0 et ultérieures incluent des fichiers JAR Kafka intégrés. Vous pouvez ignorer cette étape si vous utilisez l'une de ces versions du moteur. Pour les autres versions du moteur, procédez comme suit :

1. Téléchargez le package de dépendances correspondant à votre version de Spark :

   • Spark 3.5 : kafka-spark35-jars.zip.

   • Spark 3.4 : kafka-spark34-jars.zip.

   • Spark 3.3 : kafka-spark33-jars.zip.

2. Décompressez le package et téléchargez tous les fichiers JAR vers Object Storage Service (OSS). Cet exemple utilise kafka-spark35-jars.zip.

   hadoop fs -put /root/spark-sql-kafka-0-10_2.12-3.5.3.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/
   hadoop fs -put /root/kafka-clients-3.4.1.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/
   hadoop fs -put /root/spark-token-provider-kafka-0-10_2.12-3.5.3.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/
   hadoop fs -put /root/commons-pool2-2.11.1.jar oss://<YOUR_BUCKET>.<region>.oss-dls.aliyuncs.com/

   Vous pouvez également télécharger les fichiers via la console OSS ou par d'autres méthodes.

Étape 2 : Créer une connexion réseau

EMR Serverless Spark nécessite une connexion réseau pour accéder à Message Queue for Kafka. Pour plus d'informations, consultez Connecter EMR Serverless Spark à d'autres VPC.

Étape 3 : Préparer l'exemple d'application

Les exemples de code Scala suivants illustrent l'intégration avec Kafka. Pour davantage d'exemples, consultez Guide d'intégration Spark Structured Streaming + Kafka. Vous pouvez télécharger un fichier JAR précompilé, SparkExample-1,0-SNAPSHOT-jar-with-dependencies.jar, ou empaqueter le code vous-même. Pour la configuration pom.xml, consultez l'Annexe.

object StreamingReadKafka {

  def main(args: Array[String]): Unit = {

    import org.apache.spark.sql.SparkSession

    val servers = args(0)
    val topic = args(1)

    val spark = SparkSession.builder()
      .appName("test read kafka")
      .getOrCreate()

    import spark.implicits._

    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", servers)
      .option("subscribe", topic)
      .load()

    val query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
      .as[(String, String)]
      .writeStream
      .format("console")
      .start()

    query.awaitTermination()
  }
}

object StreamingWriteKafka {

  def main(args: Array[String]): Unit = {

    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.functions._
    import org.apache.spark.sql.streaming.Trigger

    val servers = args(0)
    val topic = args(1)
    val checkpointDir = args(2)

    val spark = SparkSession.builder()
      .appName("test write kafka")
      .getOrCreate()

    val df = spark.readStream
      .format("rate")
      .option("rowsPerSecond", "1")
      .load()
      .withColumn("key", col("value"))
      .withColumn("value", col("value"))

    val query = df
      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", servers)
      .option("topic", topic)
      .option("checkpointLocation", checkpointDir)
      .trigger(Trigger.ProcessingTime("10 seconds"))
      .start()

    query.awaitTermination()
  }
}

Étape 4 : Télécharger le package JAR

Étape 5 : Créer et exécuter une tâche de streaming

1. Sur la page EMR Serverless Spark, cliquez sur Development dans le volet de navigation de gauche.

2. Sous l'onglet Development, cliquez sur l'icône (Nouveau).

3. Dans la boîte de dialogue, saisissez un Name, sélectionnez JAR comme type sous Application (Streaming), puis cliquez sur OK.

4. Dans le coin supérieur droit, sélectionnez la file d'attente cible.

   Pour plus d'informations sur l'ajout d'une file d'attente, consultez Gérer les files d'attente de ressources.

5. Sur le nouvel onglet de développement, configurez les paramètres suivants et conservez les valeurs par défaut pour les autres paramètres. Ensuite, cliquez sur Publish.

   Paramètre	Description
   Main JAR Resource	Sélectionnez le fichier JAR que vous avez téléchargé à l'étape précédente. Cette rubrique utilise SparkExample-1,0-SNAPSHOT-jar-with-dependencies.jar à titre d'exemple.
   Engine Version	Sélectionnez une version Spark appropriée. Cette rubrique utilise esr-4.3.0 à titre d'exemple.
   Main Class	La classe principale pour le job Spark. Exemple de valeur pour la lecture depuis Kafka : org.example.StreamingReadKafka. Exemple de valeur pour l'écriture dans Kafka : org.example.StreamingWriteKafka.
   Execution Parameters	Arguments personnalisés transmis à la classe principale. Séparez plusieurs arguments par des espaces. La valeur d'exemple pour le point de terminaison Kafka dans cette rubrique est : alikafka-serverless-cn-xxxxxx-1000-vpc.alikafka.aliyuncs.com:9092. Abonnez-vous au topic Kafka. Dans cet exemple, le topic est test. Chemin d'enregistrement du checkpoint : oss://<YOUR_BUCKET_PATH>/. Ce paramètre est requis uniquement lors de l'écriture de données dans Kafka.
   Normal Network Connection	Sélectionnez la connexion réseau que vous avez créée à l'étape 2.
   Spark Configuration	Spécifiez les fichiers JAR liés à Kafka à l'aide du paramètre spark.emr.serverless.user.defined.jars. Séparez plusieurs fichiers JAR par des virgules. spark.emr.serverless.user.defined.jars oss://<YOUR_BUCKET_PATH>/commons-pool2-2.11.1.jar,oss://<YOUR_BUCKET_PATH>/kafka-clients-3.4.1.jar,oss://<YOUR_BUCKET_PATH>/spark-sql-kafka-0-10_2.12-3.5.3.jar,oss://<YOUR_BUCKET_PATH>/spark-token-provider-kafka-0-10_2.12-3.5.3.jar

6. Une fois le job publié, cliquez sur Go to O&M, puis cliquez sur START.

Étape 6 : Consulter les résultats

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.5.2</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.5.2</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.5.2</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>com.example.Main</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>