spark Streaming 自定义source数据，并实时写入到clickhouse

蔚落 2023-02-09 14:24 11阅读 0赞

1，自定义source，实时产生数据

package com.crgt.gtdata.custom
    
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.streaming.receiver.Receiver
    
    /**
      * @Author: wpp
      * @Date: 2020/5/4 23:36
      *
      */
    
    //自定义数据源 wpp
    class CustomSourceReceiver() extends Receiver[String](StorageLevel.MEMORY_ONLY) {
    
      //最初启动的时候，调用该方法，作用为：读取数据并将数据发送给spark
      override def onStart(): Unit = {
        var  idx = 0
        new Thread("Socket Receiver") {
          override def run(): Unit = {
            while (true){
              idx=idx+1
    //          store( idx.toString)
              store( idx.toString+",wpp,2020-05-18")
              Thread.sleep(100)
            }
          }
        }.start()
      }
    
    
      override def onStop(): Unit = {}
    }

2，实时朝clickhouse写入数据

package com.crgt.gtdata.custom
    
    import java.sql.Date
    import java.util.Properties
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{Row, SaveMode, SparkSession}
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    /**
      * @Author: wpp
      * @Date: 2020/5/5 0:18
      *
      */
    
    object CustomStreamSourceTest2CK2 {
    
      case class AppStayTime2(user_id:Long, user_name:String, created_date:Date)
    
    
      def main(args: Array[String]): Unit = {
        //初始化信息
        val sparkConf: SparkConf = new SparkConf().setAppName("App").setMaster("local[*]")
    
        //初始化sparkStream
        val ssc: StreamingContext = new StreamingContext(sparkConf,Seconds(5))
    
        //创建自定义的receiver的Streaming
        val lineStream: ReceiverInputDStream[String] = ssc.receiverStream(new CustomSourceReceiver())
    
        val sparkSession = SparkSession.builder()
          .appName("Test")
          .getOrCreate()
    
        import sparkSession.implicits._
        lineStream.foreachRDD(rdd => {
            val df=  rdd.map(row =>{
                    val arr = row.split(",")
                    val userId =   arr(0).toLong
                    val userName =   arr(1)
                    val createTime =   arr(2)
                    val timeStampNow = new Date(System.currentTimeMillis())
                    AppStayTime2(userId,userName,timeStampNow)
              }).toDF()
    
           df.printSchema()
           df.show(10)
    
          val connectionProperty = new Properties()
          connectionProperty.setProperty("user", "root")
          connectionProperty.setProperty("password", "root")
          connectionProperty.setProperty("databases", "crgt")
    
          df.persist(StorageLevel.MEMORY_AND_DISK_SER)
            .write.mode(SaveMode.Append)
            .option("batchsize", "5000000")
            .option("isolationLevel", "NONE") // 设置事务
            .option("numPartitions", "3") // 设置并发
            .jdbc("jdbc:clickhouse://xx.xxx.xx.xx:8123/crgt",
            "test_user",
            connectionProperty)
          df.unpersist()
    
        })
    
    
        //启动Spark StreamingContext
        ssc.start()
        ssc.awaitTermination()
    
      }
    
    
    
    }

3,pom依赖，除了spark的依赖，特别引入

<dependency>
                <groupId>ru.yandex.clickhouse</groupId>
                <artifactId>clickhouse-jdbc</artifactId>
                <version>0.2</version>
            </dependency>

4，整个的pom依赖如下：

<?xml version="1.0" encoding="UTF-8"?>
    
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    
    
    
    
    
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>gtdata-sparkmlib-video</artifactId>
        <groupId>com.crgt.gtdata</groupId>
        <version>1.0-SNAPSHOT</version>
    
        <name>gtdata-sparkmlib-video</name>
        
        <url>http://www.example.com</url>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <maven.compiler.source>1.8</maven.compiler.source>
            <maven.compiler.target>1.8</maven.compiler.target>
    
            <spark.version>2.3.4</spark.version>
            <scala.version>2.11.11</scala.version>
            <mysql.version>5.1.40</mysql.version>
            <jedis.version>2.8.0</jedis.version>
            <maven.build.timestamp.format>yyyyMMddHHmmss</maven.build.timestamp.format>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.11</version>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.specs2</groupId>
                <artifactId>specs2-core_2.11</artifactId>
                <version>4.7.0</version>
                <scope>test</scope>
            </dependency>
    
    
    
    
    
    
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>${scala.version}</version>
                
            </dependency>
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-reflect</artifactId>
                <version>${scala.version}</version>
            </dependency>
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-compiler</artifactId>
                <version>${scala.version}</version>
                
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>${spark.version}</version>
                
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.11</artifactId>
                <version>${spark.version}</version>
                
            </dependency>
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-hive_2.11</artifactId>
                <version>${spark.version}</version>
                
            </dependency>
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
                <version>${spark.version}</version>
                
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
            
            <dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
                <version>${jedis.version}</version>
            </dependency>
            <dependency>
                <groupId>com.crgt.gtdata</groupId>
                <artifactId>gtdata-common</artifactId>
                <version>1.0.0-SNAPSHOT</version>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>${mysql.version}</version>
            </dependency>
    
            <dependency>
                <groupId>joda-time</groupId>
                <artifactId>joda-time</artifactId>
                <version>2.10.3</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-slf4j-impl</artifactId>
                <version>2.11.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-api</artifactId>
                <version>2.11.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-core</artifactId>
                <version>2.11.1</version>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.62</version>
            </dependency>
            <dependency>
                <groupId>ru.yandex.clickhouse</groupId>
                <artifactId>clickhouse-jdbc</artifactId>
                <version>0.2</version>
            </dependency>
        </dependencies>
    
        <build>
            <finalName>sparkvideo</finalName>
            <resources>
                <resource>
                    <directory>src/main/resources</directory>
                </resource>
            </resources>
            <plugins>
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                            <configuration>
                                <args>
                                    <arg>-dependencyfile</arg>
                                    <arg>${project.build.directory}/.scala_dependencies</arg>
                                </args>
                            </configuration>
                        </execution>
                    </executions>
                    <configuration>
                        <scalaVersion>2.11.11</scalaVersion>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>2.4.3</version>
                    <configuration>
                    </configuration>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <filters>
                                    <filter>
                                        <artifact>*:*</artifact>
                                        <excludes>
                                            <exclude>META-INF/*.SF</exclude>
                                            <exclude>META-INF/*.DSA</exclude>
                                            <exclude>META-INF/*.RSA</exclude>
                                        </excludes>
                                    </filter>
                                </filters>
                                <artifactSet>
                                    <includes>
                                    </includes>
                                </artifactSet>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <version>3.0.0</version>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                    </configuration>
                    <executions>
                        <execution>
                        </execution>
                    </executions>
                </plugin>
    
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
    
            </plugins>
        </build>
    </project>