一、SparkStreaming概念
SparkStreaming是一个准实时的数据处理框架，支持对实时数据流进行可扩展、高吞吐量、容错的流处理，SparkStreaming可以从kafka、HDFS等中获取数据，经过SparkStreaming数据处理后保存到HDFS、数据库等。
sparkStreaming
spark streaming接收实时输入数据流，并将数据分为多个微批，然后由spark engine进行处理，批量生成最终结果流。
处理流程
二、基本操作
2.1初始化StreamingContext
Durations指定接收数据的延迟时间，多久触发一次job
SparkConf conf = new SparkConf().setMaster("local").setAppName("alarmCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
2.2基本操作
1：streamingcontext.start() 开始接受数据
2：streamingContext.stop() 停止
2.3注意的点
1:上下文启动后，不能重新设置或添加新的流式计算
2:一个JVM进程中只能有一个StreamingContext 存活
2.4DStream
DStream是离散数据流，是由一系列RDD组成的序列
1:每个InputDStream对应一个接收器（文件流不需要接收器），一个接收器也只接受一个单一的数据流，但是SparkStreaming应用中可以创建多个输入流
2:每个接收器占用一个核，应用程序的核数要大于接收器数量，如果小于数据将无法全部梳理
三、从kafka中读取数据
通过KafkaUtils从kafka读取数据，读取数据有两种方式，createDstream和createDirectStream。
3.1：createDstream：基于Receiver的方式
1: kafka数据持续被运行在Spark workers/executors 中的Kafka Receiver接受，这种方式使用的是kafka的高阶用户API
2：接受到的数据存储在Spark workers/executors内存以及WAL(Write Ahead Logs), 在数据持久化到日志后，kafka接收器才会更新zookeeper中的offset
3：接受到的数据信息及WAL位置信息被可靠存储，失败时用于重新读取数据。
createDstream读取数流程
3.2：createDirectStream 直接读取方式
这种方式下需要自行管理offset，可以通过checkpoint或者数据库方式管理
1.png
SparkStreaming
public class SparkStreaming {
private static String CHECKPOINT_DIR = "/Users/dbq/Documents/checkpoint";
public static void main(String[] args) throws InterruptedException {
//初始化StreamingContext
SparkConf conf = new SparkConf().setMaster("local").setAppName("alarmCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
jssc.checkpoint(CHECKPOINT_DIR);
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("metadata.broker.list", "172.*.*.6:9092,172.*.*.7:9092,172.*.*.8:9092");
kafkaParams.put("bootstrap.servers", "172.*.*.6:9092,172.*.*.7:9092,172.*.*.8:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "alarmGroup");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", true);
Collection<String> topics = Arrays.asList("alarmTopic");
JavaInputDStream<ConsumerRecord<String, String>> messages =
KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
);
JavaDStream<String> lines = messages.map((Function<ConsumerRecord<String, String>, String>) record -> record.value());
lines.foreachRDD((VoidFunction<JavaRDD<String>>) record -> {
List<String> list = record.collect();
for (int i = 0; i < list.size(); i++) {
writeToFile(list.get(i));
}
});
lines.print();
jssc.start();
jssc.awaitTermination();
System.out.println("----------------end");
}
//将结果写入到文件，也可以写入到MongoDB或者HDFS等
private synchronized static void writeToFile(String content) {
String fileName = "/Users/dbq/Documents/result.txt";
FileWriter writer = null;
try {
writer = new FileWriter(fileName, true);
writer.write(content + " \r\n");
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (writer != null) {
writer.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
Kafka的集成
生产者配置类
public class KafkaProducerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String broker;
@Value("${spring.kafka.producer.acks}")
private String acks;
@Value("${spring.kafka.producer.retries}")
private Integer retries;
@Value("${spring.kafka.producer.batch-size}")
private Integer batchSize;
@Value("${spring.kafka.producer.buffer-memory}")
private long bufferMemory;
public Map<String, Object> getConfig() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
props.put(ProducerConfig.ACKS_CONFIG, acks);
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
}
Kafka生产者
@Component
public class Producer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void send(Message message) {
kafkaTemplate.send("alarmTopic", JSONObject.toJSONString(message));
}
}
配置kafkaTemplate
@Component
public class PushMessageConfig {
@Autowired
private PushProducerListener producerListener;
@Autowired
private KafkaProducerConfig producerConfig;
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
@SuppressWarnings({ "unchecked", "rawtypes" })
ProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(producerConfig.getConfig());
KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(factory, true);
kafkaTemplate.setProducerListener(producerListener);
kafkaTemplate.setDefaultTopic("alarmTopic");
return kafkaTemplate;
}
}
配置生产者监听
@Component
public class PushProducerListener implements ProducerListener<String, String> {
private Logger logger = LoggerFactory.getLogger(PushProducerListener.class);
@Override
public void onSuccess(String topic, Integer partition, String key, String value,
RecordMetadata recordMetadata) {
// 数据成功发送到消息队列
System.out.println("发送成功：" + value);
logger.info("onSuccess. " + key + " : " + value);
}
@Override
public void onError(String topic, Integer partition, String key, String value,
Exception exception) {
logger.error("onError. " + key + " : " + value);
logger.error("catching an error when sending data to mq.", exception);
// 发送到消息队列失败，直接在本地处理
}
@Override
public boolean isInterestedInSuccess() {
// 发送成功后回调onSuccess，false则不回调
return true;
}
}
网友评论