from pyspark import SparkContext from pyspark.sql import SQLContext from pyspark.sql.types import Row from pyspark.sql.types import StructType from pyspark.sql.types import StructField from pyspark.sql.types import StringType from pyspark.sql.types import IntegerType if __name__ == "__main__": sc = SparkContext( 'local', 'test') spark=SQLContext(sc) 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", "123").load() jdbcDF.filter(jdbcDF.age>20).collect()//检测是否连接成功 studentRDD = sc.parallelize(["3 Mary F 26","4 Tom M 23"]).map(lambda line : line.split(" ")) schema = StructType([StructField("id",IntegerType(),True),StructField("name", StringType(), True),StructField("gender", StringType(), True),StructField("age",IntegerType(), True)]) rowRDD = studentRDD.map(lambda p : Row(int(p[0]),p[1].strip(), p[2].strip(),int(p[3]))) employeeDF = spark.createDataFrame(rowRDD, schema) prop = {} prop['user'] = 'root' prop['password'] = '123' prop['driver'] = "com.mysql.jdbc.Driver" employeeDF.write.jdbc("jdbc:mysql://localhost:3306/sparktest",'employee','append', prop) jdbcDF.collect() jdbcDF.agg({"age": "max"}).show() jdbcDF.agg({"age": "sum"}).show()

### 使用Java实现Spark SQL中的数据操作 #### RDD 转换为 Dataset 在 Spark SQL 中，`RDD` 可以通过指定模式的方式转换为 `Dataset<Row>`。以下是具体的代码实现： ```java import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SQLContext; import org.apache.spark.sql.SparkSession; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; public class RddToDatasetExample { public static void main(String[] args) { // 初始化 SparkSession SparkSession spark = SparkSession.builder() .appName("RddToDatasetExample") .master("local[*]") .getOrCreate(); SQLContext sqlContext = spark.sqlContext(); // 假设有一个简单的 JavaRDD<String> JavaRDD<String> rdd = spark.sparkContext().parallelize(Arrays.asList( "Alice,25", "Bob,30" )).toJavaRDD(); // 定义 Schema 结构 StructType schema = new StructType(new StructField[]{ DataTypes.createStructField("name", DataTypes.StringType, true), DataTypes.createStructField("age", DataTypes.IntegerType, true) }); // 将 RDD 转换为 RowRDD JavaRDD<Row> rowRDD = rdd.map(record -> { String[] parts = record.split(","); return RowFactory.create(parts[0], Integer.parseInt(parts[1])); }); // 应用 Schema 到 RDD 上并创建 Dataset<Row> Dataset<Row> dataset = sqlContext.createDataFrame(rowRDD, schema); // 显示结果 dataset.show(); // 关闭 SparkSession spark.close(); } } ``` 此代码展示了如何将一个普通的 `JavaRDD<String>` 转换为带有模式的 `Dataset<Row>`。 --- #### 保存模式到外部存储 为了持久化 `Dataset<Row>` 并保留其模式信息，可以将其写入支持模式的文件格式（如 Parquet 或 ORC）。以下是一个示例： ```java // 写入 Parquet 文件 dataset.write().mode("overwrite").parquet("/path/to/output/parquet"); // 如果需要读取回 Dataset<Row> Dataset<Row> readBackDataset = spark.read().parquet("/path/to/output/parquet"); readBackDataset.printSchema(); // 查看模式 readBackDataset.show(); // 查看数据 ``` Parquet 是一种列式存储格式，能够高效压缩和处理大规模数据[^4]。 --- #### 执行 SQL 查询 一旦有了 `Dataset<Row>`，就可以注册它为临时表或视图，并使用标准 SQL 查询对其进行操作： ```java // 注册临时视图 dataset.createOrReplaceTempView("people"); // 执行 SQL 查询 Dataset<Row> result = spark.sql("SELECT name FROM people WHERE age >= 30"); // 输出查询结果 result.show(); ``` 这允许用户利用熟悉的 SQL 语法来分析数据[^1]。 --- #### 总结 以上代码片段涵盖了以下几个方面： - 如何从 `JavaRDD<String>` 构建带模式的 `Dataset<Row>`； - 如何将 `Dataset<Row>` 持久化至外部存储； - 如何通过 SQL 对 `Dataset<Row>` 进行查询。 这些技术共同构成了基于 Java 的 Spark SQL 数据操作的核心流程。 ---

使用Spark SQL建表有多种方法，以下是三种常用的方法： 方法一，使用case class定义表： 1. 首先，使用case class定义表的结构，例如： case class Emp(empno: Int, ename: String, job: String, mgr: String, hiredate: String, sal: Int, comm: String, deptno: Int) 2. 在main函数中，创建一个SparkSession： val spark = SparkSession.builder().master("local").appName("sql").getOrCreate() 3. 导入隐式转换： import spark.sqlContext.implicits._ 4. 导入数据，这里以导入emp.csv文件为例： val lines = spark.sparkContext.textFile("G:/emp.csv").map(_.split(",")) 5. 将导入的数据映射为表的结构： val allEmp = lines.map(x=>Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt)) 6. 将映射后的数据转换为DataFrame： val empDF = allEmp.toDF() 7. 打印DataFrame： println(empDF.show()) 方法二，使用SparkSession对象： 1. 在main函数中，创建一个SparkSession： val spark = SparkSession.builder().master("local").appName("sql").getOrCreate() 2. 导入隐式转换： import spark.sqlContext.implicits._ 3. 导入数据，这里以导入emp.csv文件为例： val lines = spark.sparkContext.textFile("G:/emp.csv").map(_.split(",")) 4. 定义表的结构，这里以StructType为例： val myschema = StructType(List( StructField("empno", DataTypes.IntegerType), StructField("ename", DataTypes.StringType), StructField("job", DataTypes.StringType), StructField("mgr", DataTypes.StringType), StructField("hiredate", DataTypes.StringType), StructField("sal", DataTypes.IntegerType), StructField("comm", DataTypes.StringType), StructField("deptno", DataTypes.IntegerType) )) 5. 将导入的每一行数据映射为Row： val rowRDD = lines.map(x => Row(x(0).toInt, x(1), x(2), x(3), x(4), x(5).toInt, x(6), x(7).toInt)) 6. 使用创建DataFrame的方法将Row转换为DataFrame： val df = spark.createDataFrame(rowRDD, myschema) 7. 打印DataFrame： println(df.show()) 方法三，直接读取带格式的文件（如json文件）： 1. 在main函数中，创建一个SparkSession： val spark = SparkSession.builder().master("local").appName("sql").getOrCreate() 2. 导入隐式转换： import spark.sqlContext.implicits._ 3. 直接读取一个带格式的文件，这里以读取json文件为例： val df = spark.read.json("G:/person.json") 4. 打印DataFrame： println(df.show()) 以上是使用Spark SQL建表的三种方法，具体使用哪种方法取决于你的数据格式和需求。