Spark 读取 ElasticSearch 数据优化
个别业务上,咱们只会关怀写 ElasticSearch,写也没有用到 spark-elasticsearch 组件。应用的是 ElasticSearch 原生的 bulkProcessor。
查问方面个别在 kibana 敲敲查问代码就够了。
然而某天还是须要用到这个组件,特此记录下。
第一次应用
引入 maven
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch-spark-20_2.11</artifactId>
<version>7.8.0</version>
</dependency>
写一手简略的代码,根本整个思路有了。
import org.elasticsearch.spark.sql._
// 配置信息
val spark = SparkSession.builder().appName("ES 读取数据 商品关联视频")
.config(ConfigurationOptions.ES_NODES_WAN_ONLY, "true")
.config(ConfigurationOptions.ES_NODES, "***")
.config(ConfigurationOptions.ES_PORT, "9200")
.config(ConfigurationOptions.ES_NET_HTTP_AUTH_USER, "***")
.config(ConfigurationOptions.ES_NET_HTTP_AUTH_PASS, "***")
.getOrCreate()
val query =
"""{
| "query": {
| "bool": {
| "must": [
| {
| "term": {
| "date": {
| "value": "20200816"
| }
| }
| },
| {
| "term": {
| "room_id": {
| "value": "6861399602093165326"
| }
| }
| }
| ]
| }
| }
|}""".stripMargin
val df = spark.esDF("live-product-room-alias/origin", query)
放到线上怼一波,发现分区只有一个。跑了 4.6 小时~ 数据量在 1.2 亿左右。太慢了。
第二次应用
加两个参数下来,
意思就是 ES_SCROLL_SIZE 每次能够拿 1W 条,默认只有 50 条~
ES_MAX_DOCS_PER_PARTITION 代表每个分区解决的数据量,这个很要害,加上去之后,发现 spark stage 上的工作分区变成了 126。嗯~ 要的就是这个 feel。
.config(ConfigurationOptions.ES_SCROLL_SIZE, 10000)
.config(ConfigurationOptions.ES_MAX_DOCS_PER_PARTITION, 1000000)
往线上怼一波,报错超时。看到 Duration,发现都是在 1 分钟之后超时,很可能是有某个参数在作怪。发现了 HTTP_TIME_OUT 参数
Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 23, emr-worker-8.cluster-183597, executor 4): org.elasticsearch.hadoop.rest.EsHadoopNoNodesLeftException: Connection error (check network and/or proxy settings)- all nodes failed; tried [[es-cn-mp91asj1e0007gyzj.elasticsearch.aliyuncs.com:9200]]
/** HTTP connection timeout */
String ES_HTTP_TIMEOUT = "es.http.timeout";
String ES_HTTP_TIMEOUT_DEFAULT = "1m";
第三次应用
那么扭转一下参数好了。
.config(ConfigurationOptions.ES_SCROLL_SIZE, 10000)
.config(ConfigurationOptions.ES_MAX_DOCS_PER_PARTITION, 1000000)
.config(ConfigurationOptions.ES_HTTP_TIMEOUT,"5m")
.config(ConfigurationOptions.ES_SCROLL_KEEPALIVE,"10m")
.config("es.internal.spark.sql.pushdown", value = true)
.config(ConfigurationOptions.ES_READ_SOURCE_FILTER, "room_id,date,product_id,room_ticket_count,total_user")