将SparkSQL计算结果写入Mysql中

编写代码

package com.itunic.sql


import java.util.Properties

import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by c on 2017/1/3.
  *  Spark SQL
  *  将数据写入到MySQL中
  * by me:
  * 我本沉默是关注互联网以及分享IT相关工作经验的博客,
  * 主要涵盖了操作系统运维、计算机编程、项目开发以及系统架构等经验。
  * 博客宗旨:把最实用的经验,分享给最需要的你,
  * 希望每一位来访的朋友都能有所收获!
  */
object JdbcRDD {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("JdbcRDD").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
 //通过并行化创建RDD
    val personRDD = sc.parallelize(Array("1 tom 5", "2 jerry 3", "3 kitty 6")).map(_.split(" "))
 //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
 //将RDD映射到rowRDD
    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))
 //将schema信息应用到rowRDD上
    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)
 //创建Properties存储数据库相关属性
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "root")
 //将数据追加到数据库
    personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.155.1:3306/test", "test.t_person", prop)
 //停止SparkContext
    sc.stop()
  }
}

打包并发布到spark集群

$SPARK_HOME/bin/spark-submit \
--class com.itunic.sql.JdbcRDD \
--master spark://ResourceManagerServer1:7077 \
--jars /usr/local/spark-1.5.2-bin-hadoop2.6/mysql-connector-java-5.1.35-bin.jar \
--driver-class-path /usr/local/spark-1.5.2-bin-hadoop2.6/mysql-connector-java-5.1.35-bin.jar \
/root/spark-mvn-1.0-SNAPSHOT.jar

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Golang语言社区

linux下syscall函数,SYS_gettid,SYS_tgkill

NAME syscall - 间接系统调用 SYNOPSIS #define _GNU_SOURCE ...

38915
来自专栏大数据-Hadoop、Spark

DataFrame与RDD的互操作

DataFrame Interoperating with RDDs 参考官网 http://spark.apache.org/docs/2.2.0/sql-...

2554
来自专栏一名合格java开发的自我修养

2.sparkSQL--DataFrames与RDDs的相互转换

    当已知类的Schema的时候,使用这种基于反射的方法会让代码更加简洁而且效果也很好。

773
来自专栏xingoo, 一个梦想做发明家的程序员

Spark SQL 用户自定义函数UDF、用户自定义聚合函数UDAF 教程(Java踩坑教学版)

在Spark中,也支持Hive中的自定义函数。自定义函数大致可以分为三种: UDF(User-Defined-Function),即最基本的自定义函数,类...

3938
来自专栏数据和云

Oracle 的 DBMS_SCN 修正以及 SCN 的 auto-rollover 新特性

在 Oracle 11.2.0.2 之后,随着一系列 SCN 耗尽问题的出现,很多补丁涌现出来,一个新的 Package 增加进来。 这个 Package 就是...

3676
来自专栏杨建荣的学习笔记

rman数据备份恢复学习笔记(49天)

昨天对rman数据备份恢复进行了总结, 分享一下。 常用的一些命令 list backup; list expired backup; report schem...

3476
来自专栏技术小站

mybatis useGeneratedKey与keyProperty

612
来自专栏about云

spark2 sql读取数据源编程学习样例1

问题导读 1.dataframe如何保存格式为parquet的文件? 2.在读取csv文件中,如何设置第一行为字段名? 3.dataframe保存为表如何指定b...

5056
来自专栏Spark学习技巧

Spark高级操作之json复杂和嵌套数据结构的操作二

一,准备阶段 Json格式里面有map结构和嵌套json也是很合理的。本文将举例说明如何用spark解析包含复杂的嵌套数据结构,map。现实中的例子是,一个设备...

7979
来自专栏星汉技术

原 SparkSQL语法及API

2355

扫码关注云+社区