首页 > 数据库 > Mysql教程 > Mysql入门教程 >

关于spark连接mysql数据库后读取、写入数据

2018-07-28

关于spark连接mysql数据库后读取、写入数据。数据库连接并获取数据,从数据库中获取的数据,数据库写入数据。

数据库连接并获取数据:

 JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("sparkApp").setMaster("local[5]"));
        SQLContext sqlContext = new SQLContext(sparkContext);

 Properties Properties = new Properties();
            Properties.put("user", "数据库用户名");
            Properties.put("password", "数据库密码");
            Properties.put("driver", "数据库驱动");

DataFrame DF = sqlContext.read().jdbc("数据库地址","表名",Properties).;        

DF: 从数据库中获取的数据

数据库写入数据:

       /**
         * 1、创建类型为Row的RDD
         */
        JavaRDD> logDateRdd = sparkContext.parallelize(logDate);
        JavaRDD RDD = logDateRdd.map(new Function,Row>(){
            @Override
            public Row call(List logDate) throws Exception {
                return RowFactory.create(
                        logDate.get(0),
                        logDate.get(1)
                );
            }
        });

        /**
         * 2、动态构造DataFrame的元数据。
         */
        List structFields = new ArrayList();
        structFields.add(DataTypes.createStructField("col1",DataTypes.StringType,false));
        structFields.add(DataTypes.createStructField("col2",DataTypes.StringType,true));

        //构建StructType,用于最后DataFrame元数据的描述
        StructType structType = DataTypes.createStructType(structFields);

        /**
         * 3、基于已有的元数据以及RDD来构造DataFrame
         */
        DataFrame DF = sqlContext.createDataFrame(RDD,structType);

        /**
         * 4、将数据写入到e_trade_acct_data表中
         */
        DF.write().mode("append").jdbc("数据库地址","表名","存有用户名、密码、驱动的Properties类");

sparkContext.parallelize(logDate): 将数据转成RDD

structFields : 里面的col1 、col2为数据库字段名,DateTypes 表示数据类型,数据类型要保持一致。false:表示不能为null .true表示可为null

相关文章
最新文章
热点推荐