spark streaming 用到mysql数据怎么加载?

2021-03-30 20:04发布

2条回答
小磊子
2楼 · 2021-03-31 09:36

package com.scala.my

import org.apache.spark.SparkConf
import org.apache.spark.streaming.Durations
import org.apache.spark.streaming.StreamingContext
/**
 * 
 * @author root
 * 测试步骤:
 *    1\打开h15\h16\h17\h18,启动zookeeper,再启动hadoop集群:start-all.sh,再启动mysql
 *    2\在h15上创建文件夹wordcount_checkpoint,用于docheckpoint
 *       在h5上mysql的dg数据库中创建表t_word
 *    3\启动eclipse的本程序,让他等待着
 *    4\在h15的dos窗口下输入单词,以空格分隔的单词(需要在h15上开启端口9999:#nc -lk 9999)
 *    5\查询h15上的mysql的dg数据库的t_word表是否有数据即可
 *
 * 注:建表语句
 *     mysql> show create table wordcount;  //查看表语句
CREATE TABLE   t_word (
id  int(11) NOT NULL AUTO_INCREMENT,
updated_time  timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
word varchar(255) DEFAULT NULL,
count  int(11) DEFAULT NULL,
PRIMARY KEY (id)
);
 */
 * 
 * 测试结果:通过,注意-----》第74行没有取得数据,原因在最后没有触发事件(封装事件),目前已经解决
 * 
 * sh spark-submit --master spark://de2:7077 --class 全类名 --driver-class-path /mysql-connector-java-5.1.26.jar  sparkstreaming.jar

   sh spark-submit --class com.day6.scala.my.PresistMysqlWordCount --master yarn-cluster --driver-class-path /home/spark-1.5.1-bin-hadoop2.4/lib/mysql-connector- 

    java-5.1.31-bin.jar /home/spark-1.5.1-bin-hadoop2.4/sparkstreaming.jar 



$bin/hadoop dfsadmin -safemode leave
 
也就是关闭Hadoop的安全模式,这样问题就解决了。
 */
object PresistMysqlWordCount {


  def main(args: Array[String]): Unit = {
    //获取streamingContext,并且设置每5秒切割一次rdd
//    val sc = new StreamingContext(new SparkConf().setAppName("mysqlPresist").setMaster("local[2]"), Durations.seconds(8))
    val sc = new StreamingContext(new SparkConf().setAppName("mysqlPresist").setMaster("local[2]"), Durations.seconds(8))
    //设置checkpoit缓存策略
    /**
     * 利用 checkpoint 来保留上一个窗口的状态,
     * 这样可以做到移动窗口的更新统计
     */
    sc.checkpoint("hdfs://hh15:8020/wordcount_checkpoint")
//    sc.checkpoint("hdfs://h15:8020/wordcount_checkpoint")
    //获取doc窗口或者hdfs上的words
    //    val lines=sc.textFileStream("hdfs://h15:8020/文件夹名称")  //实时监控hdfs文件夹下新增的数据
    val lines = sc.socketTextStream("hh15", 9999)
//    val lines = sc.socketTextStream("h15", 9999)
    //压扁
    val words = lines.flatMap { x => x.split(" ") }
    //map
    val paris = words.map { (_, 1) }


    //定义一个函数,用于保持状态
    val addFunc = (currValues: Seq[Int], prevValueState: Option[Int]) => {
      var newValue = prevValueState.getOrElse(0)
      for (value <- currValues) {
        newValue += value
      }
      //返回option
      Option(newValue)
      //        //通过Spark内部的reduceByKey按key规约,然后这里传入某key当前批次的Seq/List,再计算当前批次的总和
      //        val currentCount = currValues.sum
      //        // 已累加的值
      //        val previousCount = prevValueState.getOrElse(0)
      //        // 返回累加后的结果,是一个Option[Int]类型
      //        Some(currentCount + previousCount)
    }

    //updateStateByKey操作
    /**
     * updateStateByKey 解释:
     * 以DStream中的数据进行按key做reduce操作,然后对各个批次的数据进行累加
     * 在有新的数据信息进入或更新时,可以让用户保持想要的任何状。使用这个功能需要完成两步:
     * 1) 定义状态:可以是任意数据类型
     * 2) 定义状态更新函数:用一个函数指定如何使用先前的状态,从输入流中的新值更新状态。
     * 对于有状态操作,要不断的把当前和历史的时间切片的RDD累加计算,随着时间的流失,计算的数据规模会变得越来越大。
     */
    val end = paris.updateStateByKey(addFunc)


    //插入mysql数据库
   end.foreachRDD(wd => wd.foreachPartition(
      data => {
        val conn = ConnectPool.getConn("root", "1714004716", "hh15", "dg")
//        val conn = ConnectPool.getConn("root", "1714004716", "h15", "dg")
        //插入数据
//        conn.prepareStatement("insert into t_word2(word,num) values('tom',23)").executeUpdate()
        try {
          for (row <- data) {
            println("input data is " + row._1 + "  " + row._2)
            val sql = "insert into t_word2(word,num) values(" + "'" + row._1 + "'," + row._2 + ")"
            conn.prepareStatement(sql).executeUpdate()
          }
        }finally {
          conn.close()
        }
      }))
      
    //必须添加end.print(),触发封装事件
     end.print()
    //开启接收模式
    sc.start()
    //等待
    sc.awaitTermination()
    //关闭资源
    sc.stop()
  }
}


相关问题推荐

  • 回答 10

    创建test文件夹hadoop fs -mkdir /test

  • 回答 7

    Hadoop的三大核心组件分别是:1、HDFS(Hadoop Distribute File System):hadoop的数据存储工具。2、YARN(Yet Another Resource Negotiator,另一种资源协调者):Hadoop 的资源管理器。3、Hadoop MapReduce:分布式计算框架。HDFS是一个高度容错性的系统,适合部...

  • 回答 18

    hbase依靠HDFS来存储底层数据。Hadoop分布式文件系统(HDFS)为HBase提供了高可靠性的底层存储支持,HBase中的所有数据文件都存储在Hadoop HDFS文件系统上。

  • 回答 24

    HBase分布式数据库具有如下的显著特点:容量大:HBase分布式数据库中的表可以存储成千上万的行和列组成的数据。面向列:HBase是面向列的存储和权限控制,并支持独立检索。列存储,其数据在表中是按照某列存储的,根据数据动态的增加列,并且可以单独对列进行...

  • 回答 19

    解决问题的层面不一样首先,Hadoop和Apache Spark两者都是大数据框架,但是各自存在的目的不尽相同。Hadoop实质上更多是一个分布式数据基础设施: 它将巨大的数据集分派到一个由普通计算机组成的集群中的多个节点进行存储,意味着您不需要购买和维护昂贵的服务...

  • 回答 14

    1、HBase写快读慢,HBase的读取时长通常是几毫秒,而Redis的读取时长通常是几十微秒。性能相差非常大。2、HBase和Redis都支持KV类型。但是Redis支持List、Set等更丰富的类型。3、Redis支持的数据量通常受内存限制,而HBase没有这个限制,可以存储远超内存大小...

  • 回答 15

    列式存储格式是指以列为单位存储数据的数据存储格式,相比于传统的行式存储格式,它具有压缩比高、读I/O少(此处指可避免无意义的读I/O)等优点,目前被广泛应用于各种存储引擎中。对于HBase而言,它并不是一个列式存储引擎,而是列簇式存储引擎,即同一列簇中...

  • 回答 14

    一、简单理解Hadoop是一个大象:一个hadoop集群主要包含三个主要的模块:Mapreduce,hdfs,yarn。mapreduce是一个分离在合并的计算框架,注意他不是一个集群,而是一个编程框架。hdfs是一个分布式文件系统,是一个分布式集群,用于存放数据。yarn集群是负责集群...

  • 回答 12

    01 网络公开数据集02 数据报采集03 网络爬虫04 日志收集05 社会调查06 业务数据集07 埋点采集08 传感器采集09 数据交易平台10 个人数据收集

  • 回答 9

    1 Hadoop 各个目录的解释bin:Hadoop管理脚本和使用脚本所在目录, sbin目录下的脚本都是使用此目录下的脚本实现的。etc:Hadoop的所有配置文件所在的目录,所有hadoop的配置在etc/hadoop目录下include:对外提供的库的头文件lib :对外提供的动态编程库和静态...

  • 回答 4

    HDFS存储机制,包括HDFS的写入过程和读取过程两个部分: 1、写入过程:  1)客户端向namenode请求上传文件,namenode检查目标文件是否已存在,父目录是否存在。2)namenode返回是否可以上传。3)客户端请求第一个 block上传到哪几个datanode服务器上。4)nam...

  • Shuffle 发生在哪里?2021-04-28 20:11
    回答 4

    adoop核心:MapReduce原理。 MR的核心是shuffle,被称为奇迹发生的地方。 shuffle,弄乱,洗牌的意思。partition 分区,sort 排序,spill溢出,disk 磁盘下面是官方对shuffle的配图: phase 阶段,fetch 最终,merge 合并...

  • 回答 2

    Shuffle阶段分为两部分:Map端和Reduce端。一 map端shuffle过程;1-内存预排序:默认每个map有100M内存进行预排序(为了效率),超过阈值,会把内容写到磁盘;    此过程使用快速排序算法;2-根据key和reducer的数量进行分区和排序;首先根据数据所属的Parti...

  • 回答 3

    大数据时代需要1存储大量数据2快速的处理大量数据3从大量数据中进行分析 

  • Hadoop有哪几种模式?2021-04-27 20:20
    回答 3

    hadoop的四种模式。1、本地模式:本地模式就是解压源码包,不需要做任何的配置。通常用于开发调试,或者感受hadoop。2、伪分布模式:在学习当中一般都是使用这种模式,伪分布模式就是在一台机器的多个进程运行多个模块。虽然每一个模块都有相应的进程,但是却...

  • 回答 1

    进入和退出安全模式 [root@localhost bin]# ./hdfs dfsadmin -safemode enter15/08/03 07:26:24 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where ......

没有解决我的问题,去提问