文档章节

SparkSQLJDBC数据源实例

别寒
 别寒
发布于 2017/07/27 16:56
字数 437
阅读 8
收藏 0
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
博文 271
码字总数 137605
作品 0
永州
程序员
私信 提问
解密网易MySQL实例迁移高效完成背后的黑科技

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

温正湖
2016/09/30
0
0
InnoDB--------独立表空间平滑迁移

1. 背景 * InnoDB的表空间可以是共享的或独立的。如果是共享表空间,则所有的表空间都放在一个文件里:ibdata1,ibdata2..ibdataN,这种情况下,目前应该还没办法实现表空间的迁移,除非完全迁...

asd1123509133
2017/07/20
0
0
Spring 管理配置多个数据源

Spring动态配置多数据源,即在大型应用中对数据进行切分,并且采用多个数据库实例进行管理,这样可以有效提高系统的水平伸缩性。而这样的方案就会不同于常见的单一数据实例的方案,这就要程序...

鉴客
2011/10/28
26.8K
0
bbossgroups持久层框架数据源配置文件实例

bbossgroups持久层框架数据源配置文件实例,本配置包含了物理数据源stsmc的配置实例: false

bboss
2011/07/02
0
0
SpringBoot的dao层、JdbcTemplate多数据源访问实例

前言: Spring boot集成mybatis大致过程如下: (1)新建maven project,在pom.xml文件中引入相关依赖。 (2)配置文件application.yml(application.properties) (3)编写pojo实体类 (4...

wangxuwei
2017/11/01
0
0

没有更多内容

加载失败,请刷新页面

加载更多

小白带你认识netty(三)之NioEventLoop的线程(或者reactor线程)启动(一)

在第一章中,我们看关于NioEventLoopGroup的初始化,我们知道了NioEventLoopGroup对象中有一组EventLoop数组,并且数组中的每个EventLoop对象都对应一个线程FastThreadLocalThread,那么这个...

天空小小
今天
3
0
PHP动态扩展Redis模块

查看已有模块 [root@test-a ~]# /usr/local/php/bin/php -m[PHP Modules]bz2Core...zlib[Zend Modules] 下载包,解压,生成configure文件 [root@test-a ~]# cd /usr/local/src/[ro......

野雪球
今天
3
0
在Ignite中使用线性回归算法

在本系列前面的文章中,简单介绍了一下Ignite的机器学习网格,下面会趁热打铁,结合一些示例,深入介绍Ignite支持的一些机器学习算法。 如果要找合适的数据集,会发现可用的有很多,但是对于...

李玉珏
今天
4
0
Mybatis应用学习——简单使用示例

1. 传统JDBC程序中存在的问题 1. 一个简单的JDBC程序示例: public class JDBCDemo {public static void main(String[] args) {Connection con=null;PreparedStatement statemen...

江左煤郎
今天
4
0
使用JavaScript编写iOS应用业务逻辑

JSAUIKitCocoa使你可以使用JavaScript编写对性能要求不高但可能变动性很大的iOS应用的业务逻辑部分,View组件、需要多线程支持的Model等则直接使用原生对象。 编写方式与React Native相似,但...

neal01
今天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部