文档章节

SparkSQLJDBC数据源实例

别寒
 别寒
发布于 2017/07/27 16:56
字数 437
阅读 22
收藏 0

精选30+云产品,助力企业轻松上云!>>>

package cn.hhb.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by dell on 2017/7/27.
 */
public class JDBCDataSource {
    public static void main(String[] args) {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("HiveDataSource").setMaster("local")
                .set("spark.testing.memory", "2147480000");

        // 创建javasparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        // 分别将mysql中两张表的数据加载为dataframe
        Map<String, String> options = new HashMap<String, String>();
        options.put("url","jdbc:mysql://spark1:3306/testdb");
        options.put("dbtable","student_infos");
        DataFrame studentInfosDF = sqlContext.read().format("jdbc").options(options).load();

        options.put("dbtable","student_scores");
        DataFrame studentScoresDF = sqlContext.read().format("jdbc").options(options).load();

        // 将两个dataframe转换为javapairRDD,执行join操作
        JavaPairRDD<String, Tuple2<Integer, Integer>> studentsRDD =
                studentInfosDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Row row) throws Exception {
                return new Tuple2<String, Integer>(
                    row.getString(0),
                    Integer.valueOf(String.valueOf(row.getLong(1)))
                );
            }
        }).join(studentScoresDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Row row) throws Exception {
                return new Tuple2<String, Integer>(
                        String.valueOf(row.get(0)),
                        Integer.valueOf(String.valueOf(row.get(1)))
                );
            }
        }));

        // 将javapairRDD转换为javaRDD<Row>
        JavaRDD<Row> studentRowsRDD = studentsRDD.map(new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {
            @Override
            public Row call(Tuple2<String, Tuple2<Integer, Integer>> tuple) throws Exception {
                return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
            }
        });

        // 过滤出分数大于80分的数据
        JavaRDD<Row> filteredStudentRowsRDD = studentRowsRDD.filter(new Function<Row, Boolean>() {
            @Override
            public Boolean call(Row row) throws Exception {
                if (row.getInt(2) > 80){
                    return null;
                }
                return false;
            }
        });

        // 转换为dataframe
        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
        structFields.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));
        structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
        StructType structType = DataTypes.createStructType(structFields);

        // 使用动态构造的元数据,将rdd转换为dataframe
        DataFrame studentsDF = sqlContext.createDataFrame(filteredStudentRowsRDD, structType);

        Row[] rows = studentsDF.collect();

        for (Row row : rows){
            System.out.println(row);
        }

        // 将dataFrame中的数据保存到mysql表中
        studentsDF.javaRDD().foreach(new VoidFunction<Row>() {
            @Override
            public void call(Row row) throws Exception {

                String sql = "insert into good_student_infos values('"+row.getString(0)+"','"+Integer.valueOf(String.valueOf(row.getLong(1)))+"','"+Integer.valueOf(String.valueOf(row.getLong(1)))+"')";

                Class.forName("com.mysql.jdbc.Driver");
                Connection conn = null;
                Statement stmt = null;
                try {
                    conn = DriverManager.getConnection(
                            "jdbc:mysql://spark1:3306/testdb",
                            "",
                            ""
                    );
                    stmt = conn.createStatement();
                    stmt.executeUpdate(sql);
                } catch (Exception e){
                    e.printStackTrace();
                } finally {
                    if (stmt != null){
                        stmt.close();
                    }
                    if (conn != null){
                        conn.close();
                    }
                }

            }
        });


        sc.close();
    }
}

别寒
粉丝 30
博文 273
码字总数 155300
作品 0
永州
程序员
私信 提问
加载中
请先登录后再评论。
SparkSQL 通过jdbc连接Mysql(68)

package com.dt.spark.SparkApps.sql; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.util.ArrayList; import java.util.It......

三毛_2
2016/04/13
360
0
jfinal-ext3最新版本来袭

介绍 jfinal-ext3,源自jfinal-ext,jfinal-ext2,基于jfinal3.x,扩展了很多特性。 使用 特性说明 配置说明 主要就是conf/jf-app-cfg.conf(以下简称”配置文件“)的配置说明 配置文件可以...

Jobsz
2018/07/18
0
0
Mybatis源码学习之DataSource(七)_1

简述 在数据持久层中,数据源是一个非常重要的组件,其性能直接关系到整个数据持久层的性能。在实践中比较常见的第三方数据源组件有Apache Common DBCP、C3P0、Proxool等,MyBatis不仅可以集...

osc_dst9rn7q
2018/08/04
26
0
网易云 MySQL实例迁移的技术实现

欢迎访问网易云社区,了解更多网易技术产品运营经验。 我们把数据库里部分或全部 Schema和数据迁移到另一个实例的行为称为实例迁移,将导出数据的实例称为源实例,导入数据的实例称为目标实例...

jessicaiu
2018/11/27
0
0
解密网易MySQL实例迁移高效完成背后的黑科技

作者:温正湖,网易杭研院资深工程师,负责网易云数据库平台核心开发和运维工作,对MySQL、MongoDB等数据库和Linux存储领域具有深入研究。 网易蜂巢团队:为企业提供专业容器云平台,深度整合...

温正湖
2016/09/30
0
0

没有更多内容

加载失败,请刷新页面

加载更多

1Mn18Cr18N无磁护环强度高

1Mn18Cr18N高氮奥氏体不锈钢具有强韧性好、耐蚀性好、无磁等诸多优点,成为核主泵飞轮保持环材料的选择。 保持环的热套工序要求保持环材料具备优良的热膨胀性能,飞轮保持环完整性的保证要求...

无磁钢
18分钟前
10
0
比较器,Comparator与Comparable

Comparable比较器 从JDK1.2后提供了比较器的接口:Comparable接口。 public interface Comparable<T>{ /** * 实现对象的比较处理操作 * @param o 要比较的对象 * @return...

哼着我的小调调
30分钟前
9
0
以每种语言编译时,在C和C ++中都有效的代码能否产生不同的行为?

问题: C and C++ have many differences, and not all valid C code is valid C++ code. C和C ++有很多区别,并非所有有效的C代码都是有效的C ++代码。 (By "valid" I mean standard code w......

富含淀粉
34分钟前
7
0
使用getApplication()作为上下文的对话框抛出“无法添加窗口-令牌null不适用于应用程序”

问题: My Activity is trying to create an AlertDialog which requires a Context as a parameter. 我的活动试图创建一个AlertContext,它需要一个Context作为参数。 This works as expect......

法国红酒甜
今天
13
0
亚马逊测评买家号多开_可以解决这个问题嘛?_微信公众号: VMlogin中文版

对于很多亚马逊卖家来说,做亚马逊测评是并不可少的,都在为了自己的店铺能够获得更多的销售,着重培养自己产品的各项属性,以求获得一个更好的权重排名从而获得更多的曝光,但是在旺季期间亚...

竹节猫-ASOer
今天
10
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部