Structured Streaming 读取kafka 写入Neo4j-阿里云开发者社区

Structured Streaming 读取kafka 写入Neo4j

2022-11-26 254

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Structured Streaming 读取kafka 写入Neo4j

Output Modes

有几种输出模式：

Append mode (default) - 这是默认模式，其中只有添加到结果表(因为最后一个触发器)的新行将输出到接收器。这只适用于那些添加到结果表的行永远不会改变的查询。因此，此模式保证每个行只输出一次(假设容错接收器)。例如，只有select、where、map、flatMap、filter、join等查询将支持Append模式。
Complete mode - 每次触发后，整个结果表将输出到接收器。这对于聚合查询是受支持的。
Update mode - (从Spark 2.1.1起可用)只有结果表中自最后一个触发器以来更新的行将输出到接收器。

OutputSinks

1.文件接收器-----将输出存储到目录中

writeStream.format("parquet")        // 可以是"orc", "json", "csv"等等格式
    .option("path", "path/to/destination/dir").start()

2.Kafka 接收器——将输出发送到Kafka中的一个或多个Topic。

writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("topic", "updates")
    .start()

3.Foreach接收器——在输出中的记录上运行任意计算

wordCount.writeStream
      .outputMode("update")
      .foreach(new ForeachWriter[Row] {
        //插入数据, 当有重复的 key 的时候更新，
        val sql = "insert into word_count values(?, ?) on duplicate key update word=?, count=?";
        var conn: Connection = null
        // open 一般用于打开链接，返回false时则表示跳过该区的数据
        override def open(partitionId: Long, epochId: Long): Boolean = {
          Class.forName("com.mysql.jdbc.Driver")
          conn = DriverManager.getConnection("jdbc:mysql://c702:3306/ss", "root", "Root.123")
          //如果conn不为空，而且conn 没有关，则可以往外写
          conn != null && !conn.isClosed
        }
        //把数据写入到连接中
        override def process(value: Row): Unit = {
          val ps: PreparedStatement = conn.prepareStatement(sql)
          ps.setString(1,value.getString(0))
          ps.setLong(2,value.getLong(1))
          ps.setString(3,value.getString(0))
          ps.setLong(4,value.getLong(1))
          ps.execute()
          ps.close()
        }
        //关闭连接
        override def close(errorOrNull: Throwable): Unit = {
          if (conn != null && !conn.isClosed) conn.close()
        }
      })
      .start()
    query.awaitTermination()

4. ForeachSink接收器——可以实现批量写入或者实现写入多个任意位置。

默认情况下，foreachBatch仅提供至少一次写入保证。但是，您可以使用提供给函数的 batchId 作为对输出进行重复数据删除并获得精确一次保证的方法。

wordCount.writeStream
      .outputMode("complete")
      .foreachBatch((df,batchId) => {
        df.persist()//持久化
        df.write.mode("overwrite")//如果有则覆盖
          .jdbc("jdbc:mysql://c701:3306/ss","word_count",props)//写到jdbc
        df.write.mode("overwrite").json("./foreachBatch")//写到json
        df.unpersist()
      })
      .start()
    query.awaitTermination()
  }

5.内存接收器(同样用于调试)——输出作为内存中的表存储在内存中。这两种模式都支持Append和Complete输出模式。整个输出被收集并存储在Driver程序的内存中，这应该用于在低数据量上进行调试。因此，请谨慎使用。

writeStream
    .format("memory")
    .queryName("tableName")
    .start()

Event time must be defined on a window or a timestamp 时间时间必须为时间戳
Sorting is not supported on streaming DataFrames/Datasets, unless it is on aggregated DataFrame/Dataset in Complete output mode;排序在流式处理的时候不支持
输出模式必须是append或update. 在输出模式是complete的时候(必须有聚合), 要求每次输出所有的聚合结果. 我们使用 watermark 的目的是丢弃一些过时聚合数据, 所以complete模式使用wartermark无效也无意义
在输出模式是append时, 必须设置 watermask 才能使用聚合操作. 其实, watermask 定义了 append 模式中何时输出聚合聚合结果(状态), 并清理过期状态
在输出模式是update时, watermask 主要用于过滤过期数据并及时清理过期状态.
watermask 会在处理当前批次数据时更新, 并且会在处理下一个批次数据时生效使用. 但如果节点发送故障, 则可能延迟若干批次生效
withWatermark 必须使用与聚合操作中的时间戳列是同一列.df.withWatermark(“time”, “1 min”).groupBy(“time2”).count() 无效
withWatermark 必须在聚合之前调用f.groupBy(“time”).count().withWatermark(“time”, “1 min”) 无效

Structured Streaming 写入Neo4j demo

1: 简单数据流向

2: 引入pom.xml 相关依赖

<dependency>
    <groupId>org.neo4j</groupId>
    <artifactId>neo4j</artifactId>
    <version>3.4.18</version>
</dependency>
<dependency>
    <groupId>org.neo4j</groupId>
    <artifactId>neo4j-spatial</artifactId>
    <version>0.25.5-neo4j-3.4.1</version>
</dependency>
<dependency>
    <groupId>neo4j-contrib</groupId>
    <artifactId>neo4j-spark-connector</artifactId>
    <version>2.4.1-M1</version>
</dependency>

3: 样例数据：

{"change":[{"kind":"insert","schema":"public","table":"sample","columnnames":["xzqhbm","qylx","ybbq","ybbqdm","ybpssj","fbl","dtyxmc","whq","zjzt","sfyh","sfzd","sfyy","area","length","xcenter","ycenter","ybzxyxq","ybzdyxq","dem","slope","geometry"],"columntypes":["character varying(255)","character varying(255)","character varying(255)","character varying(255)","date","numeric","character varying(255)","bigint","bigint","bigint","bigint","bigint","double precision","double precision","numeric","numeric","date","date","bigint","bigint","public.geometry"],"columnvalues":["370303","平原","乔灌果园","YD010100","2020-05-12",1,"GF-2",0,0,0,0,0,0.00940095134656,0,0,0,null,null,0,0,"0106000000010000000103000000010000001500000090138B24A9265E4090596931FC454240881F4397B5265E400021E1D4F745424060C2D3F1B9265E402057AE1AF845424068E585DEBC265E4058B9B4DAF6454240A09FEB60BC265E4050131A81EB454240E84044A7BD269E4028CB3343EB45424020BFC614BD265E406809D346E1454240B06AA148BD265E4000F9D6EADF454240807380ACBC265E40B8E9315FD8454240C8F9D769B9265E40A8ADA1D9D745424068382007BA265E4038AD4E1ADF454240380E5AE7B6265E4040A64984E1454240F0AB13E6A9265E403877A5E6E445424050FC8C36A7265E40D871155BE645424068D89587A6265E40680C1980E745424000C0390FA6265E40482D6882EA45424008A88F11A7265E40B0963FA2F345424048CE2A82AB265E4098FEDC3FF2454240E80FFDE2AB265E40F026E831F545424018FC146CA8265E40580A4D21F645424090138B24A9265E4090596931FC454290","kafka2neo4j"]}]}

4: 核心代码如下：

import org.apache.spark.sql.{DataFrame, ForeachWriter, Row, SparkSession}
import org.json.{JSONArray, JSONObject}
import org.neo4j.driver.v1.{AuthTokens, Driver, GraphDatabase, Session}
/**
 * @Description TODO
 * @Data 2020/10/21 10:16
 * @Created by mdz
 */
object GetDataForeach {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("kafka2Delta")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._
    val kafkaDF: DataFrame =
      spark
        .readStream //readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "192.168.0.219:9092,192.168.0.220:9092,192.168.0.221:9092")
        .option("subscribe", "db02_topic")
        .option("startingOffsets", "latest") //earliest  //latest
        .option("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        .option("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        .load()
    val resDF: DataFrame = kafkaDF
      .select(col("value")
        .cast("String").as("json"))
      .filter(col("json").contains("kafka2neo4j"))
      .filter(col("json").contains("insert"))
      .distinct()
    val query = resDF.writeStream
      .foreach(new ForeachWriter[Row] {
        var leaderPath: String = _
        var driver: Driver = _
        var session: Session = _
        override def open(partitionId: Long, epochId: Long): Boolean = {
          println("開始創建連接.....")
          leaderPath = "bolt://localhost:27687"
          driver = GraphDatabase.driver(leaderPath, AuthTokens.basic("neo4j", "neo4j"))
          session = driver.session()
          driver != null && session != null
        }
        override def process(value: Row): Unit = {
          println("開始執行任務....")
          val jsonData: JSONObject =new JSONObject(value.mkString)
          val jsonArray: JSONArray = jsonData.getJSONArray("change")
          val valuedata = jsonArray.get(0).toString
          val json = new JSONObject(valuedata)
          val columntypes: JSONArray = json.getJSONArray("columntypes")
          val columnvalues: JSONArray = json.getJSONArray("columnvalues")
          val columnnames: JSONArray = json.getJSONArray("columnnames")
          var cql = "CREATE (n {"
          for(i <- 0 until columnnames.length()){
            if(columntypes.get(i).toString.contains("character")||columntypes.get(i).toString.contains("geometry")||columntypes.get(i).toString.contains("date")){
              cql = cql+columnnames.get(i) + ":'" +columnvalues.get(i) + "',"
            }else{
              cql = cql+columnnames.get(i) + ":" +columnvalues.get(i) + ","
            }
          }
          cql  =cql.substring(0,cql.length-1) +  "})"
          println(cql)
          session.run(cql)
        }
        override def close(errorOrNull: Throwable): Unit = {
          println("關閉連接.....")
          session.close()
          driver.close()
        }
      }).start()
    query.awaitTermination()
  }
}

Structured Streaming 读取kafka 写入Neo4j

Output Modes

OutputSinks

热门文章

最新文章

相关课程

相关电子书

相关实验场景