Apache Kafka 和Spark的实时数据管道

<groupId>io.keisan.knowledgebase.spark.streaming</groupId>
<artifactId>keisan-spark-kafka</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<!-- Project Information -->
<name>Spark Kafka Direct Integration</name>
<description>Keisan Knowledgebase Spark Streaming Project - Kafka Direct Integration</description>
<url>https://www.keisan.io/knowledgebase</url>
<organization>
    <name>Keisan Ltd</name>
    <url>https://www.keisan.io</url>
</organization>
<developers>
    <developer>
        <id>jillur.quddus</id>
        <name>Jillur Quddus</name>
        <email>contactus@keisan.io</email>
        <url>https://www.keisan.io</url>
        <organization>Keisan Ltd</organization>
        <organizationUrl>https://www.keisan.io</organizationUrl>
        <roles>
            <role>Lead Engineer</role>
            <role>Data Scientist</role>
        </roles>
        <timezone>Europe/London</timezone>
    </developer>
</developers>
<!-- Repositories -->
<repositories>
    <!-- Confluent Repository for KafkaAvroDecoder -->
    <repository>
        <id>confluent</id>
        <url>http://packages.confluent.io/maven/</url>
    </repository>
</repositories>
<!-- Properties -->
<properties>
    <apache.avro.version>1.8.1</apache.avro.version>
    <apache.spark.core.2.11.version>2.1.0</apache.spark.core.2.11.version>
    <apache.spark.streaming.2.11.version>2.1.0</apache.spark.streaming.2.11.version>
    <apache.spark.streaming.kafka-0-8_2.11.version>2.1.0</apache.spark.streaming.kafka-0-8_2.11.version>
    <confluent.kafka.avro.serializer.version>3.2.1</confluent.kafka.avro.serializer.version>
    <jdk.version>1.8</jdk.version>
    <maven.plugins.maven-assembly-plugin.version>3.0.0</maven.plugins.maven-assembly-plugin.version>
    <maven.plugins.maven-compiler-plugin.version>3.6.1</maven.plugins.maven-compiler-plugin.version>
    <output.directory>/keisan/knowledgebase/spark/jars</output.directory>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<!-- Dependencies -->
<dependencies>
    <!-- Apache Avro -->
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>${apache.avro.version}</version>
    </dependency>
    <!-- Apache Spark -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>${apache.spark.core.2.11.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>${apache.spark.streaming.2.11.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
        <version>${apache.spark.streaming.kafka-0-8_2.11.version}</version>
    </dependency>
    <!-- Confluent Kafka Avro Serializer -->
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>${confluent.kafka.avro.serializer.version}</version>
    </dependency>
</dependencies>
<!-- Build -->
<build>
    <!-- Plugins -->
    <plugins>
        <!-- Maven Compiler: Compile the Sources of the Project -->
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>${maven.plugins.maven-compiler-plugin.version}</version>
            <configuration>
                <source>${jdk.version}</source>
                <target>${jdk.version}</target>
            </configuration>
        </plugin>
        <!-- Maven Assembly: Aggregate project output with its dependencies -->
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>${maven.plugins.maven-assembly-plugin.version}</version>
            <configuration>
                <!-- Final JAR Filename -->
                <finalName>keisan-spark-kafka-${project.version}</finalName>
                <!-- Include all Project Dependencies -->
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
                <!-- JAR with dependencies Output Target Directory -->
                <outputDirectory>${output.directory}</outputDirectory>
            </configuration>
            <executions>
                <!-- Bind the assembly to the package phase -->
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

我们现在准备开始开发我们的Spark Streaming Application。要创建输入DStream，我们需要导入_KafkaUtils_并使用Spark Streaming Context创建直接流，Kafka Broker正在侦听的主机名和端口以及我们要从中使用消息的Kafka主题。<br />**简单的纯文本处理**<br />在下面的示例中，我们只是使用Kafka的推文，将值解码为Strings并将它们输出到Spark Executor Standard Out。
```java
package io.keisan.knowledgebase.spark.streaming.kafka;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import kafka.serializer.StringDecoder;
/**
 * Example Spark Streaming Application with Kafka Direct Integration
 *
 * Periodically query Kafka for the latest offsets in each Topic and Partition.
 * Consume the tweets using a String Decoder and display them in the console.
 *
 * Usage: StreamingKafkaDirectStringDecoder
 *  broker: The hostname and port at which the Kafka Broker is listening
 *
 * @author jillur.quddus
 * @version 0.0.1
 *
 */
public class StreamingKafkaDirectStringDecoder {
    public static void main(String[] args) throws InterruptedException {
        if ( args.length != 1 ) {
            System.err.println("Usage: StreamingKafkaDirectStringDecoder <broker>");
            System.exit(1);
        }
        // Create a Java Streaming Context with a Batch Interval of 5 seconds
        SparkConf conf = new SparkConf().setAppName("Kafka Direct String Decoder");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
        // Specify the Kafka Broker Options and set of Topics
        String broker = args[0];
        Map<String, String> kafkaParameters = new HashMap<String, String>();
        kafkaParameters.put("metadata.broker.list", broker);
        Set<String> topics = Collections.singleton("twitter");
        // Create an input DStream using KafkaUtils and simple plain-text message processing
        JavaPairInputDStream<String, String> kafkaDirectStream = KafkaUtils.createDirectStream(jssc,
                String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParameters, topics);
        kafkaDirectStream.foreachRDD(rdd -> {
            rdd.foreach(record -> System.out.println(record._2));
        });
        // Start the computation
        jssc.start();
        // Wait for the computation to terminate
        jssc.awaitTermination();
    }
}

Avro处理
请注意，我们在数据管道中使用的Flume Twitter Source实际上将推文转换为Avro格式，并将这些Avro消息发送到下游。Avro是一个基于数据序列化模式和语言中立的系统，它使用JSON来声明数据结构和模式。您可以使用返回Bytes原始数组的DefaultDecoder，然后使用Avro的二进制解码器或Confluent的KafkaAvroDecoder解码它，以接收带有Avro记录的消息，而不是使用简单地将值解码为上面的Streaming应用程序中的字符串的StringDecoder。他们的价值观
运行Spark Streaming应用程序
我们现在准备运行我们的Spark Streaming Application！使用Maven构建胖 JAR并将其提交给Spark集群。因为出于本文的目的，我们仅使用单节点Spark集群进行开发和调试，所以在客户端模式下部署我们的应用程序是有意义的。

# Submit our Spark Streaming Application
cd $SPARK_HOME
bin/spark-submit --class io.keisan.knowledgebase.spark.streaming.kafka.StreamingKafkaDirectStringDecoder --deploy-mode client /keisan/knowledgebase/spark/jars/keisan-spark-kafka-0.0.1-SNAPSHOT-jar-with-dependencies.jar <Kafka Broker Hostname:Port>
# Launch the Flume Agent using the Twitter Source and Kafka Channel
cd $FLUME_HOME
bin/flume-ng agent --name TwitterAgent --conf conf --conf-file conf/flume-twitter.conf

如果您现在检查Spark Executor输出，您应该能够看到从近乎实时地从Kafka流向Spark的推文！
我们现在已经成功开发了一种近实时，高吞吐量，可靠且容错的数据流水线。Apache Flume用于收集发布到Apache Kafka的基于事件的数据（在我们的示例中为推文）。然后，Apache Spark用于使用来自Apache Kafka的数据并执行接近实时的数据处理。显然，简单地摄取和打印数据绝不会展示Apache Spark的功能！因此，在我的下一篇文章中，我将讨论如何在Spark中基于事件的数据流实时构建预测模型和实时数据分析。