sparksql jdbc数据源

用的本地模式,pom.xml中添加了mysql驱动包,mysql已经开启,写入的时候发现用format("jdbc").save()的方式发现会有does not allow create table as select的异常,于是去官方文档上发现了使用jdbc()的方式,测试

正常,说明下Properties是java.util.Properties

java

 public class Demo {

     private static SparkSession session = SparkSession.builder().appName("demo").master("local").getOrCreate();

     public static void main(String[] args) {

         Map<String, String> options = new HashMap<>();

         options.put("url", "jdbc:mysql://127.0.0.1:3306/studentmanage");

         options.put("driver", "com.mysql.jdbc.Driver");

         options.put("dbtable", "studentmanage.admin");

         options.put("user", "root");

         options.put("password", "root");

         // 读取

         Dataset<Row> dataset = session.read().format("jdbc").options(options).load();

         dataset.show();

         // 创建数据

         List<Row> list = new ArrayList<Row>();

         Row row1 = RowFactory.create("tele", "123", "male", "China", 1, "admin");

         Row row2 = RowFactory.create("wyc", "123", "male", "China", 1, "admin");

         Row row3 = RowFactory.create("xxx", "123", "male", "China", 1, "admin");

         list.add(row1);

         list.add(row2);

         list.add(row3);

         // 写入

         StructType schema = DataTypes

                 .createStructType(Arrays.asList(DataTypes.createStructField("name", DataTypes.StringType, false),

                         DataTypes.createStructField("pwd", DataTypes.StringType, false),

                         DataTypes.createStructField("sex", DataTypes.StringType, false),

                         DataTypes.createStructField("nation", DataTypes.StringType, false),

                         DataTypes.createStructField("status", DataTypes.IntegerType, false),

                         DataTypes.createStructField("type", DataTypes.StringType, false)));

         Dataset<Row> ds = session.createDataFrame(list, schema);

         Properties connectionProperties = new Properties();

         connectionProperties.put("user", "root");

         connectionProperties.put("password", "root");

         // 也可以对dataset进行遍历使用原生的jdbc或者dbutils等进行写入

         ds.write().mode(SaveMode.Append).jdbc("jdbc:mysql://127.0.0.1:3306/studentmanage", "admin",

                 connectionProperties);

         session.stop();

     }

 }

scala

 object Demo {

   def main(args: Array[String]): Unit = {

     val session = SparkSession.builder().appName("demo").master("local").getOrCreate()

     val options = Map[String, String](

       ("url", "jdbc:mysql://127.0.0.1:3306/studentmanage"),

       ("driver", "com.mysql.jdbc.Driver"),

       ("dbtable", "studentmanage.admin"),

       ("user", "root"),

       ("password", "root"))

     //读取

     val df = session.read.options(options).format("jdbc").load()

     df.show()

     //写入

     val arrBuffer = Array(Row("yeye", "123", "male", "us", 1, "admin")).toBuffer

     val schema = DataTypes.createStructType(Array(

       StructField("name", DataTypes.StringType, false),

       StructField("pwd", DataTypes.StringType, false),

       StructField("sex", DataTypes.StringType, false),

       StructField("nation", DataTypes.StringType, false),

       StructField("status", DataTypes.IntegerType, false),

       StructField("type", DataTypes.StringType, false)))

     val result = session.createDataFrame(arrBuffer, schema)

     val properties = new Properties

     properties.put("user", "root")

     properties.put("password", "root")

     result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://127.0.0.1:3306/studentmanage", "admin", properties)

     session.stop

   }

 }

巴特西

sparksql jdbc数据源

最新文章

热门文章