利用Spark进行对账demo

//利用cogroup 处理分隔符的文件
import org.apache.spark.{ SparkContext, SparkConf }
import java.sql.DriverManager
object HandleGroup extends App{

  val beginTime = System.currentTimeMillis()
  //引用spark
  val conf = new SparkConf().setAppName("Simple Application").setMaster("local[*]")
  val sc = new SparkContext(conf)

  //获取文件
  val txt1 = sc.textFile("i:/1/fg1.txt").map(line =>line.split( "\\|"))
  val txt2 = sc.textFile("i:/1/fg2.txt").map(line =>line.split( "\\|"))

  val results = txt1.map{v =>(v(0)+v(1)+v(5),v)  }.cogroup(
    txt2.map{v =>(v(0)+v(1)+v(5),v)}).filter {
    case (k, (v1, v2)) => !(v1.nonEmpty&&v2.nonEmpty) }

  //JDBC预处理
  val driver = "oracle.jdbc.driver.OracleDriver"
  val url = "jdbc:oracle:thin:@"
  val username = ""
  val password = ""
  Class.forName(driver)
  val connection = DriverManager.getConnection(url, username, password)
  val statement = connection.createStatement()

  //遍历结果集
  results.foreach{case (k,(v1,v2)) =>
    v1.foreach(a =>runSql(a(0),a(1),a(5)))
    v2.foreach(a => runSql(a(0),a(1),a(5)))
  }

  //执行sql语句
  def runSql(v1:String,v2:String,v3:String): Unit =
    statement.addBatch("insert into nidaye( v1, v2, v3) values('"+v1+"','"+v2+"','"+v3+"')")

  //提交
  val resultSet = statement.executeBatch()
  connection.close()
  println(System.currentTimeMillis() - beginTime)
}

 

//利用SPARKSQL 处理定长分隔的文件
import java.sql.DriverManager

import org.apache.spark.rdd.JdbcRDD

import org.apache.spark.{ SparkContext, SparkConf }
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{ IntegerType, StringType, StructField, StructType }

// 导入 Row.
import org.apache.spark.sql.Row;

// 导入 Spark SQL 数据类型
import org.apache.spark.sql.types.{ StructType, StructField, StringType };
object HandleLength extends App{
  val beginTime = System.currentTimeMillis()
  //引用spark
  val conf = new SparkConf().setAppName("Simple Application").setMaster("local[*]")
  val sc = new SparkContext(conf)

  //JDBC预处理
  val driver = "oracle.jdbc.driver.OracleDriver"
  val url = "jdbc:oracle:thin"
  val username = ""
  val password = ""
  Class.forName(driver)
  val connection = DriverManager.getConnection(url, username, password)
  val statement = connection.createStatement()

  //创建sparkSql上下文
  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
   //获取数据源
  val people = sc.textFile("i:/1/dc1g_1_temp.txt")
  val dog = sc.textFile("i:/1/dc1g_2_temp.txt")
  val t1 = (System.currentTimeMillis() - beginTime)
  //定义schema
  val schemaResources = "v1 v2 v3 v4 v5 v6 v7 v8 v9 v10 v11 v12 v13 v14 v15 v16 v17 v18 v19 v20 v21 v22 v23 v24 v25 v26"
  //val schemaResources = "v1 v2 v3"

  // 转化成结构化类型
  val Resources1 = StructType(schemaResources.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
  val Resources2 = StructType(schemaResources.split(" ").map(fieldName => StructField(fieldName, StringType, true)))


  val t2 = (System.currentTimeMillis() - beginTime)
  //将数据源进行重新匹配
  val rowRDD1 = people.map(p => splitString(p)).map(p => Row(p(0).trim, p(1).trim, p(2).trim, p(3).trim, p(4).trim, p(5).trim, p(6).trim, p(7).trim, p(8).trim, p(9).trim
    , p(10).trim, p(11).trim, p(12).trim, p(13).trim, p(14).trim, p(15).trim, p(16).trim, p(17).trim, p(18).trim, p(19).trim, p(20).trim, p(21).trim, p(22).trim, p(23).trim, p(24).trim, p(25).trim))
  val rowRDD2 = dog.map(p => splitString(p)).map(p => Row(p(0).trim, p(1).trim, p(2).trim, p(3).trim, p(4).trim, p(5).trim, p(6).trim, p(7).trim, p(8).trim, p(9).trim
    , p(10).trim, p(11).trim, p(12).trim, p(13).trim, p(14).trim, p(15).trim, p(16).trim, p(17).trim, p(18).trim, p(19).trim, p(20).trim, p(21).trim, p(22).trim, p(23).trim, p(24).trim, p(25).trim))

  val t3 = (System.currentTimeMillis() - beginTime)
  // 将模板和RDD进行数据结构化
  val peopleDataFrame = sqlContext.createDataFrame(rowRDD1, Resources1)
  val dogDataFrame = sqlContext.createDataFrame(rowRDD2, Resources2)
  val t4 = (System.currentTimeMillis() - beginTime)
  // 注册DataFrames为表。

  peopleDataFrame.registerTempTable("res1")
  dogDataFrame.registerTempTable("res2")
  import org.apache.spark.storage.StorageLevel
  //持久化
  peopleDataFrame.persist(StorageLevel.DISK_ONLY  )
  dogDataFrame.persist(StorageLevel.DISK_ONLY  )

  val t5 = (System.currentTimeMillis() - beginTime)
  // SQL语句可以通过使用由sqlContext提供的SQL方法运行。
  //使用126
  val resultsAB = sqlContext.sql("SELECT t1.v1,t1.v2,t1.v3 FROM res1 t1 except SELECT t2.v1,t2.v2,t2.v3 FROM res2 t2  ")
  val t8 = (System.currentTimeMillis() - beginTime)
  val resultsBA = sqlContext.sql("SELECT t2.v1,t2.v2,t2.v3 FROM res2 t2 except SELECT t1.v1,t1.v2,t1.v3 FROM res1 t1  ")

  val t6 = (System.currentTimeMillis() - beginTime)
  // SQL查询的结果是DataFrames支持所有的正常的RDD操作。
  //val sum = results

  //println("-----单边A未匹配的总数："+resultsAB.count()+" ----------")
  //println("-----单边B未匹配的总数："+resultsBA.count()+" ----------")
  //results.map(t => "v1: " + t(0)+" v2: " + t(1) + " v3:"+ t(2)).collect().foreach(println)
  val t7 = (System.currentTimeMillis() - beginTime)
  //入库

  resultsAB.collect().foreach(t => runSql(t(0).toString+"-AB",t(1).toString+"-AB",t(2).toString+"-AB"))

  resultsBA.collect().foreach(t => runSql(t(0).toString+"-BA",t(1).toString+"-BA",t(2).toString+"-BA"))
  val t9 = (System.currentTimeMillis() - beginTime)

  //按长度截取
  def splitString(str:String):Array[String] ={
    val loop ="19,45,65,92,111,118,123,133,143,153,158,165,172,179,184,191,197,200,206,223,226,232,266,271,276,282,289"
    val arrLoop = loop.split(",")
    val arrL = arrLoop.map(a => Integer.parseInt(a))
    val arr = new Array[String](26)

    var i:Int = -1
    for(a <- arrL if(i!=25)){
      i = i+1
      arr(i)  = str.substring(arrL(i),arrL(i+1))
    }
    arr(25) =str.substring(arrL(25))

    return arr
  }


  //执行sql语句
  def runSql(v1:String,v2:String,v3:String): Unit =

    statement.addBatch("insert into nidaye( v1, v2, v3) values('"+v1+"','"+v2+"','"+v3+"')")
    //println("SQL执行完成： "+resultSet)


  val resultSet = statement.executeBatch()
  connection.close()
  println("总耗时："+(System.currentTimeMillis() - beginTime))
  println(t1+" "+t2+" "+t3+" "+t4+" "+t5+" "+t6+" "+t7+" "+t8+" "+t9)
}

 

//利用SPARKSQL 处理分隔符的文件
import java.sql.DriverManager

import org.apache.spark.rdd.JdbcRDD

import org.apache.spark.{ SparkContext, SparkConf }
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{ IntegerType, StringType, StructField, StructType }

// 导入 Row.
import org.apache.spark.sql.Row;

// 导入 Spark SQL 数据类型
import org.apache.spark.sql.types.{ StructType, StructField, StringType };

object Handle extends App {
  val beginTime = System.currentTimeMillis()
  //引用spark
  val conf = new SparkConf().setAppName("Simple Application").setMaster("local[*]")
  val sc = new SparkContext(conf)

  //JDBC预处理
  val driver = "oracle.jdbc.driver.OracleDriver"
  val url = "jdbc:oracle:thin"
  val username = ""
  val password = ""
  Class.forName(driver)
  val connection = DriverManager.getConnection(url, username, password)
  val statement = connection.createStatement()

  //创建sparkSql上下文
  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    //获取数据源
  val people = sc.textFile("i:/1/dc1g_1_temp.txt")
  val dog = sc.textFile("i:/1/dc1g_2_temp.txt")

  //定义schema
  val schemaResources = "v0 v1 v2 v3 v4 v5 v6 v7 v8 v9 v10 v11 v12 v13 v14 v15 v16 v17 v18 v19 v20 v21 v22 v23 v24 v25 v26"
  //val schemaResources = "v1 v2 v3"

  // 转化成结构化类型
  val Resources1 = StructType(schemaResources.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
  val Resources2 = StructType(schemaResources.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

  //将数据源进行重新匹配
  val rowRDD1 = people.map(_.split("\\|")).map(p => Row(p(0).trim, p(1).trim, p(2).trim, p(3).trim, p(4).trim, p(5).trim, p(6).trim, p(7).trim, p(8).trim, p(9).trim
    , p(10).trim, p(11).trim, p(12).trim, p(13).trim, p(14).trim, p(15).trim, p(16).trim, p(17).trim, p(18).trim, p(19).trim, p(20).trim, p(21).trim, p(22).trim, p(23).trim, p(24).trim, p(25).trim, p(26).trim))
  val rowRDD2 = dog.map(_.split("\\|")).map(p => Row(p(0).trim, p(1).trim, p(2).trim, p(3).trim, p(4).trim, p(5).trim, p(6).trim, p(7).trim, p(8).trim, p(9).trim
    , p(10).trim, p(11).trim, p(12).trim, p(13).trim, p(14).trim, p(15).trim, p(16).trim, p(17).trim, p(18).trim, p(19).trim, p(20).trim, p(21).trim, p(22).trim, p(23).trim, p(24).trim, p(25).trim, p(26).trim))

  // 将模板和RDD进行数据结构化
  val peopleDataFrame = sqlContext.createDataFrame(rowRDD1, Resources1)
  val dogDataFrame = sqlContext.createDataFrame(rowRDD2, Resources2)
  import org.apache.spark.storage.StorageLevel
  //持久化
  peopleDataFrame.persist(StorageLevel.DISK_ONLY  )
  dogDataFrame.persist(StorageLevel.DISK_ONLY  )

  //peopleDataFrame.filter(dogDataFrame)
  // 注册DataFrames为表。
  peopleDataFrame.registerTempTable("res1")
  dogDataFrame.registerTempTable("res2")




  // SQL语句可以通过使用由sqlContext提供的SQL方法运行。
  //使用126

  val resultsAB = sqlContext.sql("SELECT t1.v0,t1.v1,t1.v2,t1.v3 FROM res1 t1 left outer join res2 t2 on t1.v0=t2.v0 where t2.v1 is null ")
  val t8 = (System.currentTimeMillis() - beginTime)
  val resultsBA = sqlContext.sql("SELECT t2.v0,t2.v1,t2.v2,t2.v3 FROM res2 t2 left outer join res1 t1 on t1.v0=t2.v0 where t1.v1 is null ")

  // SQL查询的结果是DataFrames支持所有的正常的RDD操作。
  resultsAB.collect().foreach(t => runSql(t(1).toString+"-AB",t(2).toString+"-AB",t(3).toString+"-AB"))

  resultsBA.collect().foreach(t => runSql(t(1).toString+"-BA",t(2).toString+"-BA",t(3).toString+"-BA"))

  //执行sql语句
  def runSql(v1:String,v2:String,v3:String): Unit =

    statement.addBatch("insert into nidaye( v1, v2, v3) values('"+v1+"','"+v2+"','"+v3+"')")

  val resultSet = statement.executeBatch()
  connection.close()
  println(System.currentTimeMillis() - beginTime)
}