SparkSQL之数据写入

比眉伴天荒 2023-01-02 01:17 290阅读 0赞

# SparkSQL之数据写入 #

package com.spark.sparksql
    
    import java.sql.{ Connection, DriverManager, PreparedStatement}
    import java.util.Properties
    
    import org.apache.spark.sql.{ Row, SaveMode, SparkSession}
    import org.junit.Test
    
    
    class fileWrite { 
      
      val spark = SparkSession.builder().master("local[4]")
        .appName("test").getOrCreate()
      
      import spark.implicits._
      
      /** * 数据写入也是全部采用一套Api： df/ds.write * 数据写入有两种方式： * 1、df/ds.write * .mode --设置写入模式 * .format() --设置写入格式[csv/parquent/json/text/jdbc] * .option() --设置写入需要的参数 * .save(path) --设置数据写入路径 * 2、df/ds.write.mode.json() * df/ds.write.mode.parquet() * df/ds.write.mode.csv() * df/ds.write.mode.jdbc() * df/ds.write.mode.text() */
      
      @Test
      def writeText():Unit={ 
      
        val df1 = spark.createDataset(List(("zhangsan",20),("lisi",30),("wangwu",40),("zhoaliu",50)))
      
        val df2 = df1.toDF("name","age")
      
        /** * 写入模式： * SaveMode.Overwrite:写入路径如果存在，则覆盖里面的数据【常用，数据写入到hdfs的时候使用】 * SaveMode.Append:写入路径如果存在，则将当前数据追加进去【常用，数据写入到mysql的时候使用】 * SaveMode.Ignore：写入路径如果存在，不做任何操作 * SaveMode.ErrorIfExists: 写入路径如果存在则报错 */
      
        //以json格式写入数据
        //第二种写入方式
        df1.write.mode(SaveMode.Overwrite).json("output/json")
        //第一种写入方式
        df1.write.mode(SaveMode.Append).format("json").save("output/json")
      
      
        val df3 = spark.createDataset(List("hello java","hello spark"))
        //以文本格式写入（写入时只能有一个列，数据类型为string）
        //第一种方式
        df3.write.mode(SaveMode.Overwrite).format("text").save("output/text")
        //第二种方式
        df3.write.mode(SaveMode.Append).text("output/text")
        
      }
      
      
      @Test
      def writeParquet:Unit={ 
      
        val df1 = spark.createDataset(List(("zhangsan",20),("lisi",30),("wangwu",40),("赵流",50)))
        
        df1.write.mode(SaveMode.Overwrite).format("parquet").save("output/parquet")
        df1.write.mode(SaveMode.Overwrite).parquet("output/parquet")
      }
      
      
      @Test
      def writeCsv:Unit={ 
        val df1 = spark.createDataset(List((1,"zhangsan",20),(2,"yyyy",30),(3,"wangwu",40),(4,"ssss",50))).toDF("id","name","age")
      
        df1.write.mode(SaveMode.Overwrite).format("csv").option("seq","#").option("header","true")
          .save("output/csv")
      
        df1.write.mode(SaveMode.Append).option("seq","*").option("header","true").csv("output/csv")
        
      }
      
      
      @Test
      def writeJDBC:Unit={ 
      
        val df1 = spark.createDataset(List((1,"yyy",24),(2,"uuu",8),(3,"ttt",40),(4,"ssss",50))).toDF("id","name","age")
        
        //第一种方式
        df1.write.format("jdbc")
          .option("url","jdbc:mysql://hadoop102:3306/test")
          .option("dbtable","test")
          .option("user","root")
          .option("password","123321")
          .mode(SaveMode.Append)
          //.save()
        
        
        val props = new Properties()
        props.setProperty("user","root")
        props.setProperty("password","123321")
        
        //第二种方式
        try { 
          df1.write.mode(SaveMode.Overwrite).jdbc("jdbc:mysql://hadoop102:3306/test", "test", props)
        }catch { 
          case e=>
        }
        
        //上面两种方式，如果采用SaveMode.Append模式，如果表中有主键可能会出现主键冲突，所以此时需要自己使用foreachPartition进行数据写入
        df1.foreachPartition((it:Iterator[Row])=>{ 
      
          var connection: Connection = null
          var statement: PreparedStatement = null
           
            try{ 
              connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test","root","123321")
              statement = connection.prepareStatement("update test set name=?,age=? where id =?")
        
              it.foreach(row=>{ 
                val name = row.getAs[String]("name")
                val age = row.getAs[Int]("age")
                val id = row.getAs[Int]("id")
                
                statement.setInt(3,id)
                statement.setString(1,name)
                statement.setInt(2,age)
                
                statement.execute()
              })
            }catch { 
              case e:Exception=>
                e.printStackTrace()
            }finally { 
              if (statement != null)
                statement.close()
              if(connection!=null)
                connection.close()
            }
        })
      
    }