streaming读取kafka数据再保存到es7.7.0代码及踩过的坑

超、凢脫俗 2023-02-12 12:29 2阅读 0赞

### 代码 ###

import org.apache.kafka.clients.consumer.ConsumerConfig
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.DStream
    import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.elasticsearch.spark.rdd.EsSpark
    
    
    
    object StreamingWriteES {
    
      def main(args: Array[String]): Unit = {
        // offset保存路径
        // val checkpointPath = "D:\\hadoop\\checkpoint\\kafka-direct"
    
        val session = SparkSession.builder()
          //.master("local[*]")
          .appName("kafkaStreamingWriteES")
          .config("xpack.security.user","elastic:elastic123")//x-pack认证安全机制的账号密码
          .config("es.net.http.auth.user", "elastic")//es的账号
          .config("es.net.http.auth.pass", "elastic123")//es的密码
          .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .getOrCreate()
        session.sparkContext.setLogLevel("WARN")
    
        val ssc = new StreamingContext(session.sparkContext, Seconds(1))
        //  ssc.checkpoint(checkpointPath)
    
        val bootstrapServers = "xxxx:9092,xxxx:9092,xxxx:9092"
        val groupId = "test-consumer-group3"
        val topicName = "ecar-photo-gps" //
        val maxPoll = 1000
    
        val kafkaParams = Map(
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers,
          ConsumerConfig.GROUP_ID_CONFIG -> groupId,
          ConsumerConfig.MAX_POLL_RECORDS_CONFIG -> maxPoll.toString,
          ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
          ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
        )
    
        val kafkaTopicDS = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferBrokers,
          ConsumerStrategies.Subscribe[String, String](Set(topicName), kafkaParams))
    
        val dStream: DStream[String] = kafkaTopicDS.map(_.value)
          .flatMap(_.split(" "))
    
    
        val esConf = Map(
         "es.nodes.wan.only" -> "true",
          "es.nodes" -> "xxx",//单机版es
          "es.port" -> "9200",
          "es.resource" -> "test/ecar_data" //es索引
        )
    
        dStream.foreachRDD(rdd => {
          EsSpark.saveJsonToEs(rdd, esConf)
        })
    
        ssc.start()
        ssc.awaitTermination()
    
      }
    }

### maven ###

<dependencies>
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.11</artifactId>
                <version>2.1.1</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
                <version>2.1.1</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>2.1.1</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
                <version>2.1.1</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.httpcomponents</groupId>
                <artifactId>httpclient</artifactId>
                <version>4.4.1</version>
            </dependency>
    
            
            <dependency>
                <groupId>commons-httpclient</groupId>
                <artifactId>commons-httpclient</artifactId>
                <version>3.1</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.httpcomponents</groupId>
                <artifactId>httpmime</artifactId>
                <version>4.3.6</version>
            </dependency>
    
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>2.11.8</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.elasticsearch</groupId>
                <artifactId>elasticsearch</artifactId>
                <version>7.7.0</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.elasticsearch</groupId>
                <artifactId>elasticsearch-hadoop</artifactId>
                <version>7.7.0</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.elasticsearch.client</groupId>
                <artifactId>transport</artifactId>
                <version>7.7.0</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.56</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-pool2</artifactId>
                <version>2.4.2</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.postgresql</groupId>
                <artifactId>postgresql</artifactId>
                <version>42.1.4</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-configuration2</artifactId>
                <version>2.2</version>
            </dependency>
            <dependency>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
                <version>1.2.17</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>jcl-over-slf4j</artifactId>
                <version>1.7.22</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.22</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.7.22</version>
            </dependency>
    
            
        </dependencies>

### 错误1： ###

### Exception in thread "streaming-job-executor-0" java.lang.NoClassDefFoundError: org/apache/commons/httpclient/protocol/ProtocolSocketFactory ###

### **解决办法：** ###

### [去这个网址里面找相对应的依赖：https://mvnrepository.com/][https_mvnrepository.com] ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTgwNDA0OQ_size_16_color_FFFFFF_t_70][]

### **错误2：** ###

### 报什么elasticsearch-spark-20\_2.11依赖和elasticsearch-hadoop依赖同时存在 ###

### **解决办法：** ###

### 只留elasticsearch-hadoop依赖就可以了 ###

### 注意：打jar包放到spark上运行的时候又会报错缺少elasticsearch-spark-20\_2.11这个包，然后我将它放在submit的脚本中，如下： ###

### \--jars  /sparkapp/cxb/kafka\_write\_es/elasticsearch-spark-20\_2.11-7.7.0.jar ###

#!/bin/bash
    
    nohup spark-submit --master yarn --deploy-mode cluster --class com.emg.es.StreamingWriteES --name es --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 30 --jars /sparkapp/cxb/kafka_write_es/elasticsearch-spark-20_2.11-7.7.0.jar /sparkapp/cxb/kafka_write_es/es.jar

### 错误3：  ###

{
      "error" : "Content-Type header [application/x-www-form-urlencoded] is not supported",
      "status" : 406
    }

### 解决办法： ###

### 此原因是公司安装ES时开启了x-pack安全机制，进行严格的内容类型检查，严格检查内容类型也可以作为防止跨站点请求伪造攻击的一层保护。 ###

.config("xpack.security.user","elastic:elastic123")//x-pack认证安全机制的账号密码

### 使用命令行进行读写的时候不用设置，如下：指定一个json文件传输到es中 ###

curl -H "Content-Type: application/json"  -XPOST localhost:9200/people/shakespeare/_bulk?pretty  --data-binary test.json

### 错误4： ###

20/05/26 15:35:39 ERROR rest.NetworkClient: Node [emg4032:9200] failed (The host did not accept the connection within timeout of 60000 ms); no other nodes left - aborting...

### 解决办法： ###

### 因为我的es和spark是装在不同的机器上，然后是用sprak on yarn 运行的， ###

### 所以 es机器中 /etc/hosts   --->   映射spark所在机器IP ###

### 然后再设置  spark机器中 /etc/hosts   -->   映射es所在机器IP ###

[https_mvnrepository.com]: https://mvnrepository.com/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTgwNDA0OQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200526114055141.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTgwNDA0OQ==,size_16,color_FFFFFF,t_70