import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

val peopleRDD = spark.sparkContext.textFile("file:///opt/software/employee.txt")

val schemaString = "id name age"

val fields = schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, nullable = true))

val schema = StructType(fields)

val rowRDD = peopleRDD.map(_.split(",")).map(attributes => Row(attributes(0), attributes(1).trim, attributes(2).trim))

val peopleDF = spark.createDataFrame(rowRDD, schema)

peopleDF.createOrReplaceTempView("people")

val results = spark.sql("SELECT id,name,age FROM people")

results.map(attributes => "id: " + attributes(0)+","+"name:"+attributes(1)+","+"age:"+attributes(2)).show(false)

id

name

gender

Age

1

Alice

F

22

2

John

M

25

id

name

gender

age

3

Mary

F

26

4

Tom

M

23

package spark.core.exper05

/**
* @ClassName TestMysql.java
* @author 趙浩博
* @version 1.0.0
* @Description TODO
* @createTime 2022年01月19日 14:31:00
*/
import java.util.Properties
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}
object TestMysql {
def main(args: Array[String]) {
val spark=SparkSession.builder().getOrCreate()
val employeeRDD = spark.sparkContext.parallelize(Array("3 Mary F 26","4 Tom M 23")).map(_.split(" "))
val schema = StructType(List(StructField("id", IntegerType, true),StructField("name", StringType, true),StructField("gender", StringType, true),StructField("age", IntegerType, true)))
val rowRDD = employeeRDD.map(p => Row(p(0).toInt,p(1).trim, p(2).trim,p(3).toInt))
val employeeDF = spark.createDataFrame(rowRDD, schema)
val prop = new Properties()
prop.put("user", "root")
prop.put("password", "1229")
prop.put("driver","com.mysql.cj.jdbc.Driver")
employeeDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest", "sparktest.employee", prop)
val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/sparktest").option("driver","com.mysql.jdbc.Driver").option("dbtable","employee").option("user","root").option("password", "1229").load()
jdbcDF.agg("age" -> "max", "age" -> "sum")
}
}