一、应用场景二、环境说明三、实验步骤四、思考
一、应用场景针对实时处理的数据需要及时能够搜索出来时，可以选择elasticsearch来支持这一业务。当然还可以选择其他的内存数据库，如redis。而elasticsearch除了强大的全文索引能力外，还支持分布式存储，可以将其作为分布式计算框架的底座，用于存储热数据或者温数据等。常见的组合方式有elasticsearch与sparkstreaming以及flink进行配合共同完成实时或流式处理的场景。
在本次案例中，我们将演示将商品信息数据通过nc发送，由sparkstreaming通过socket读取实时数据流并对其进行简单处理，最后写入到es中。
二、环境说明- 数据生产工具：nc （本案例安装在虚拟机中）数据处理框架：sparkstreaming,版本2.12数据存储框架：elasticsearch开发工具IDEAes集群：3台节点 （本案例安装在虚拟机中）http通讯端口：9200
各种环境搭建，如es、idea等请自行搭建
启动idea，创建maven项目
添加依赖至pom.xml中
org.apache.spark spark-core_2.12 3.0.0 org.apache.spark spark-streaming_2.12 3.0.0 org.elasticsearch elasticsearch 7.8.0 org.elasticsearch.client elasticsearch-rest-high-level-client 7.8.0 org.apache.logging.log4j log4j-api 2.8.2 org.apache.logging.log4j log4j-core 2.8.2
添加scala框架的支持，或者在pom.xml中自行添加
编写sparkstreaming客户端程序，完成代码如下：
package com.suben.es.sstream
import org.apache.http.HttpHost
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import java.util.Date
object SparkStreamingESTest {
val INDEX_NAME = "stream"
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreamingOperateES")
val streamingContext = new StreamingContext(sparkConf, Seconds(3))
// 读取TCP端口数据，即nc发送过来的数据
val inputDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("hadoop003", 12345, StorageLevel.MEMORY_ONLY)
// 对数据进行处理
inputDStream.foreachRDD(rdd => {
println(">>>>>>>>>>>>" + new Date())
rdd.foreach(data => {
println("==========接收到的数据>>>>>>>>>>>>", data)
if (data != null && !"".equals(data)) {
// 将数据写入到elasticsearch索引为shopping中去
// 0. 处理数据,假如输入数据格式如下：9,小米手机,手机,2008.0,http://www.suben.com/xm.jpg
val splits = data.split(",")
// 选择第一个元素作为id
val id = splits(0)
// 1. 编写es客户端代码
val client = new RestHighLevelClient(RestClient.builder(new HttpHost("hadoop002", 9200, "http")))
// 2. 创建请求对象
val request = new IndexRequest();
// 3. 设置索引和id
request.index(INDEX_NAME).id(id)
// 4. 将splits数组中的数据转成json字符串
val productJson =
s"""
| { "id":"${splits(0)}",
| "title":"${splits(1)}",
| "category":"${splits(2)}",
| "price":"${splits(3)}",
| "images":"${splits(4)}" }
|""".stripMargin;
println("===productJson===", productJson)
// 6. 将jsonObject封装到request中
request.source(productJson, XContentType.JSON)
// 7. 请求添加文档
val response = client.index(request, RequestOptions.DEFAULT)
System.out.println("_index:" + response.getIndex());
System.out.println("_id:" + response.getId());
System.out.println("_result:" + response.getResult());
client.close()
}
})
})
// 启动程序
streamingContext.start()
// 等待执行结果
streamingContext.awaitTermination()
}
}
启动nc，命令如下：
nc -l -p 12345
启动sparkstreaming客户端程序
在nc中添加测试数据，回车，观察es中有无数据被写入
10,[10]华为手机,手机,3008.0,http://www.suben.com/hw.jpg
11,[11]华为手机,手机,3008.0,http://www.suben.com/hw.jpg
12,[12]华为手机,手机,3008.0,http://www.suben.com/hw.jpg
13,[13]华为手机,手机,3008.0,http://www.suben.com/hw.jpg
14,[14]华为手机,手机,5008.0,http://www.suben.com/hw.jpg
在chrome浏览器插件elasticsearch-head中数据被成功写入es中：
上述实验仅仅实现了简单的将通过nc数据逐条或者多条输入，是手动的方式操作的。假如是真实业务场景，如电商系统一直源源不断地产生数据，如需模拟这种场景，又该如何实现呢？欢迎大家在评论区给我言，我们一起讨论，相互学习吧！
