关于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

本文标题：关于spark连接MYSQL数据库后读取、写入数据
本文地址：http://www.q0738.com/mysql/1248.html

• RDBMS术语
• 创建后表的修改：修改列
• 操作MYSQL数据库：更新表中的数据