Spark Streaming企业级开发入门实战

作者: LY丶Smile | 来源:发表于2021-05-28 19:28 被阅读0次

写在前头

本文主要是Spark Streaming学习过程中写的一些示例代码，如有bug，请留言指正，不胜感激。

1. 包含功能

Spark Streaming checkpoint的getOrCreate方法（详见源码中的RealDemo）
Spark Streaming集成Kafka
Spark Streaming与Hbase
Spark Streaming与Redis
Spark Streaming与MySQL

2. 环境

集群：CDH6.3.1
Spark版本：2.4.0-cdh6.3.1
Hbase版本：2.1.0-cdh6.3.1

3. 源码地址，点击访问GitHub

4. 一个最接近真实业务需求的示例--具体见源码中的RealDemo

一、理论基础

1.1 Spark Streaming是什么

Spark Streaming是构建在Spark上的实时计算框架，扩展了Spark流式大数据的处理能力
Spark Streaming接收实时流的数据，并根据一定的时间间隔拆分成一批批的数据，这些批数据在Spark内核对应一个RDD实例，然后进行处理

1.2 应用场景

Spark Streaming可以从多种数据源（kafka、RocketMQ、Flume、HDFS等）获取数据。

Spark Streaming严格意义上来讲是的伪实时，延迟为秒级，本质上还是批处理，适合处理实时性要求不苛刻的准实时场景。如果希望收到一条数据就立马进行处理，那么Storm、Flink更符合要求。

1.3 开发要点

Spark Streaming本质上还是一个批处理，所以数据的输出最好还是批量输出，这样效率会高一点。
资源的获取（数据库连接等）最好是rdd.foreachPartition，在每个partition上进行资源的获取及释放。因为RDD是分布式的数据集，分布式服务是涉及到网络传输的，而网络的传输就涉及到序列化，操作略显复杂。而每个partition可以当成一个单体服务，按照普通的开发思维进行数据处理，复杂度会低不少。
Spark Streaming的开发应该是简单的，如果开始长篇大论，可能是在重复造轮子（个人感触，不一定准确），最好是找一些业界成熟的解决方案。

二、Spark Streaming集成Kafka

2.1 依赖引入

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.4.0</version>
</dependency>

2.2 接收kafka消息

JavaInputDStream<ConsumerRecord<String, String>> messages =
                    KafkaUtils.createDirectStream(
                            jsc,
                            LocationStrategies.PreferConsistent(),
                            ConsumerStrategies.Subscribe(
                                    MqUtils.getTopics("sparkRedisDemo"),
                                    MqUtils.buildKafkaParams("127.0.0.1:9092", "sparkRedis")));

// 解析消息体，生成一个新的DStream
JavaDStream<String> lines = messages.map(ConsumerRecord::value);

三、Spark Streaming将数据存入Hbase

Spark Streaming中使用Hbase有两种常用的方式，如下

Spark内置的saveAsNewAPIHadoopDataset，封装好的方法，用起来比较简单。比较适合一个RDD对应一个输出的场景。
手动创建Client，直接使用HbaseClient，虽然多写点代码，但是灵活性要高很多。可以针对每条数据进行差异化处理及输出。

3.1 依赖引入

<!--for hbase-->
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-server</artifactId>
  <version>${hbase.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-client</artifactId>
  <version>${hbase.version}</version>
</dependency>

3.2 示例代码

3.2.1 方法一：Spark内置方法

public static void saveBySpark(JavaPairRDD<String, String> rdd, String tableName) {

    Configuration config = HBaseConfiguration.create();
    config.set("hbase.mapred.outputtable", tableName);
    config.set("mapreduce.job.outputformat.class", "org.apache.hadoop.hbase.mapreduce.TableOutputFormat");

    JavaPairRDD<ImmutableBytesWritable, Put> hbasePuts = rdd.mapToPair(line -> {
        Put put = new Put(Bytes.toBytes(line._1));
        Map<String, Object> values = JSONObject.parseObject(line._2);
        values.forEach((k, v) -> {
            try {
                put.addColumn(HbaseTableConsts.BYTES_COL_FAMILY, Bytes.toBytes(k), ObjectUtils.toByte(v));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        return new Tuple2<>(new ImmutableBytesWritable(), put);
    });
    hbasePuts.saveAsNewAPIHadoopDataset(config);
}

调用方式

JavaDStream<String> lines = messages.map(ConsumerRecord::value);
// 设置缓存，内存不足时会刷到硬盘
lines.persist(StorageLevel.MEMORY_AND_DISK());

// 数据解析为 rowkey:dataValue
JavaPairDStream<String, String> data2Hbase = lines.mapToPair(d -> {
    MqRequestData data = MqUtils.parseMessage(d);
    return new Tuple2<>(HbaseUtils.buildRowkey(data), JSON.toJSONString(data.getValues()));
});

// 存储到Hbase
data2Hbase.foreachRDD(rdd -> HbaseUtils.saveBySpark(rdd, "spark_demo"));

3.2.2 方法二：HbaseClient

/**
 * 自己建立client的方式操作Hbase
 * - 更灵活
 */
public static void saveByClient(Connection conn, String tableName, List<Put> puts) throws IOException {
    try (Table table = conn.getTable(TableName.valueOf(tableName))) {
        table.put(puts);
    }
}

调用方式

private static void saveToHbaseByClient(JavaDStream<String> lines) {
        lines.foreachRDD(rdd -> rdd.foreachPartition(p -> {
            try (Connection conn = ConnectionFactory.createConnection(HBaseConfiguration.create())) {
                Map<String, List<Put>> dataMap = new HashMap<>(8);
                while (p.hasNext()) {
                    MqRequestData data = JSONObject.parseObject(p.next(), MqRequestData.class);
                    String tableName = HbaseUtils.getTableName(data);
                    Put put = HbaseUtils.buildPut(data, HbaseUtils.buildRowkey(data));
                    List<Put> puts = dataMap.getOrDefault(tableName, new ArrayList<>());
                    puts.add(put);
                    dataMap.put(tableName, puts);
                }

                dataMap.forEach((k, v) -> {
                    try {
                        HbaseUtils.saveByClient(conn, k, v);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            }
        }));
    }

四、Spark Streaming与Redis

4.1 依赖引入

<dependency>
  <groupId>com.redislabs</groupId>
  <artifactId>spark-redis_2.11</artifactId>
  <version>2.4.2</version>
</dependency>

4.2 示例代码

4.2.1 方式1：使用Spark-Redis封装好的方法

/**
 * Spark 操作redis
 * @author smile
 */
public class SparkRedisDemo {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("sparkRedisDemo")
                .setMaster("yarn")
                .set("spark.redis.host", "127.0.0.1")
                .set("spark.redis.port", "6379")
                .set("spark.redis.auth", "123456")
                .set("spark.redis.db", "2");

        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1));
        jsc.sparkContext().setLogLevel("ERROR");

        RedisConfig redisConfig = new RedisConfig(new RedisEndpoint(conf));
        ReadWriteConfig readWriteConfig = ReadWriteConfig.fromSparkConf(conf);
        RedisContext redisContext = new RedisContext(jsc.ssc().sc());

        try {
            JavaInputDStream<ConsumerRecord<String, String>> messages =
                    KafkaUtils.createDirectStream(
                            jsc,
                            LocationStrategies.PreferConsistent(),
                            ConsumerStrategies.Subscribe(
                                    MqUtils.getTopics("sparkRedisDemo"),
                                    MqUtils.buildKafkaParams("127.0.0.1:9092", "sparkRedis")));

            // map 将函数应用于每个RDD的每个元素，返回值是新的RDD
            JavaDStream<String> lines = messages.map(ConsumerRecord::value);
            // 设置缓存，内存不足时会刷到硬盘
            lines.persist(StorageLevel.MEMORY_AND_DISK());

            JavaPairDStream<String, Seq<String>> data2Redis = lines.mapToPair(d -> {
                MqRequestData data = MqUtils.parseMessage(d);
                String key = StringUtils.join(data.getProjectCode(), "", data.getDeviceCode());
                List<String> s = Collections.singletonList(d);
                // java List 转 scala Seq
                Seq<String> seq = JavaConverters.asScalaIteratorConverter(s.iterator()).asScala().toSeq();
                return new Tuple2<>(key, seq);
            });

            // 将数据写入redis list
            data2Redis.foreachRDD(rdd -> redisContext.toRedisLISTs(rdd.rdd(), 0, redisConfig, readWriteConfig));

            jsc.start();
            jsc.awaitTermination();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

4.2.2 一种更灵活地方式

直接通过ConnectionPool获取jedis实例，然后就可以实现任何自己想实现的操作了。但注意最好是一个partition用一个连接，防止连接数过多占用一些不必要的资源

/**
 * 另外一种更灵活获取Jedis连接的方法
 * - 获取Jedis实例，然后可以灵活地进行各种操作
 */
private static Jedis getJedis(SparkConf conf) {
    return ConnectionPool.connect(new RedisEndpoint(conf));
}

五、Spark Streaming与MySQL

5.1 依赖引入

使用Druid连接池

<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>druid</artifactId>
  <version>1.2.6</version>
</dependency>

5.2 连接池

/**
 * 数据库连接池
 * @author smile
 */
public class JdbcConnectionPool {

    private static DataSource ds;

    static {
        try {
            Properties prop = new Properties();
            InputStream resourceAsStream = JdbcConnectionPool.class.getClassLoader().getResourceAsStream("jdbc.properties");
            prop.load(resourceAsStream);
            ds = DruidDataSourceFactory.createDataSource(prop);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取连接
     */
    public static Connection getConnection() throws SQLException {
        return ds.getConnection();

    }

    /**
     * 关闭数据库的资源
     */
    public static void close(Connection conn, PreparedStatement ps, ResultSet rs) {
        if (null != conn) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (null != ps) {
            try {
                ps.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (null != rs) {
            try {
                rs.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}

5.3 调用实例

public static void exportToMysql(JavaDStream<MqRequestData> lines) {
    lines.foreachRDD(rdd -> rdd.foreachPartition(p -> {
        Connection conn = JdbcConnectionPool.getConnection();
        String sql = "INSERT INTO mysql_demo(id,name,created_time) VALUES(?,?,NOW())";
        PreparedStatement ps = conn.prepareStatement(sql);
        conn.setAutoCommit(false);
        while (p.hasNext()) {
            MqRequestData data = p.next();
            ps.setString(1, String.valueOf(data.getValues().get("id")));
            ps.setString(2, String.valueOf(data.getValues().get("name")));
            ps.addBatch();
        }
        ps.executeBatch();
        conn.commit();
        ps.clearBatch();
        JdbcConnectionPool.close(conn, ps, null);
    }));
}

附录：工具类

1. 消息队列工具类

public static Map<String, Object> buildKafkaParams(String servers, String groupId) {
        // 构建kafka参数map
        Map<String, Object> kafkaParams = new HashMap<>(8);
        //Kafka服务监听端口
        kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // earliest
        kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 以下注释掉的三行为kafka安全验证配置-SASL
        // kafkaParams.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name);
        // kafkaParams.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        // kafkaParams.put(SaslConfigs.SASL_JAAS_CONFIG,
        //        "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";");
        return kafkaParams;
    }


public static Collection<String> getTopics(String kafkaTopics) {
    Collection<String> topics = new HashSet<>();
    Collections.addAll(topics, kafkaTopics.split(","));
    return topics;
}

public static MqRequestData parseMessage(String message) {
    return JSONObject.parseObject(message, MqRequestData.class);
}

2. hbase操作工具类

/**
 * hbase操作 工具类
 * @author smile
 */
public class HbaseUtils {

    /**
     * Spark内置的方式操作hbase
     */
    public static void saveBySpark(JavaPairRDD<String, String> rdd, String tableName) {

        Configuration config = HBaseConfiguration.create();
        config.set("hbase.mapred.outputtable", tableName);
        config.set("mapreduce.job.outputformat.class", "org.apache.hadoop.hbase.mapreduce.TableOutputFormat");

        JavaPairRDD<ImmutableBytesWritable, Put> hbasePuts = rdd.mapToPair(line -> {
            Put put = new Put(Bytes.toBytes(line._1));
            Map<String, Object> values = JSONObject.parseObject(line._2);
            values.forEach((k, v) -> {
                try {
                    put.addColumn(HbaseTableConsts.BYTES_COL_FAMILY, Bytes.toBytes(k), ObjectUtils.toByte(v));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
            return new Tuple2<>(new ImmutableBytesWritable(), put);
        });
        hbasePuts.saveAsNewAPIHadoopDataset(config);
    }

    /**
     * 自己建立client的方式操作Hbase
     * - 更灵活
     */
    public static void saveByClient(Connection conn, String tableName, List<Put> puts) throws IOException {
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            table.put(puts);
        }
    }

    public static String buildRowkey(MqRequestData data) {
        return StringUtils.join(data.getProjectCode(), HbaseTableConsts.ROWKEY_SEP, data.getDeviceCode(), HbaseTableConsts.ROWKEY_SEP, data.getCreated());
    }

    public static String getTableName(MqRequestData data) {
        return "demo";
    }


    public static Put buildPut(MqRequestData data, String rowkey) {
        Put put = new Put(Bytes.toBytes(rowkey));
        data.getValues().forEach((k, v) -> {
            try {
                put.addColumn(HbaseTableConsts.BYTES_COL_FAMILY, Bytes.toBytes(k), ObjectUtils.toByte(v));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        return put;
    }
}