我记得刚开始接触大数据的时候最绕不开的一个坎就是Spark。而Spark里面入门门槛最低、后面工作里用得最多的就是Spark SQL的DataFrame和DataSet。很多教程要么直接贴官网文档要么甩一堆概念字符串、列、Schema、编码器……看完一头雾水开IDEA还是不知道从哪下手。这篇文章我就按我自己上手时候的顺序来直接用IDEA搭一个Maven工程从SparkSession初始化开始把DataFrame和DataSet的创建、转换、常见坑全部过一遍全程代码可复制。适合刚学完Scala基础、准备进坑Spark的人也适合用Java写Spark但想对比一下两种API差异的朋友。1. 动手前的整体思考为什么DataFrame与DataSet值得专门学1.1 DataFrame与DataSet的身世从RDD到更友好的抽象很多教程喜欢从RDD开始讲导致很多人以为RDD是Spark的“本体”。实际上RDD确实是Spark最初的抽象但它有两个很明显的痛点一是没有schema结构数据是什么字段、什么类型全靠写代码的人自己记二是性能优化全靠程序员手动做了能做的优化也有限。后来Spark SQL出现了带来了DataFrame。DataFrame是什么你可以把DataFrame看成一张分布式的表有列名、有列类型像数据库表一样。它的底层仍然是RDD但外表换了一套更友好的API。更重要的是Spark对DataFrame做了Catalyst优化器简单说就是Spark会自动帮你想“这个查询怎么写更快”很多性能优化在内部就完成了不需要你操心。DataSet呢它是在DataFrame基础上又加了一层类型约束。DataFrame的每一行是Row类型这种是弱类型的列的类型错了要到运行期才发现。而DataSet每一行是一个具体的case class对象比如People、Order这些编译期就能检查出类型对不对。换句话说DataFrame DataSet[Row]这是理解两者的一个关键公式。1.2 两个API各自的脾气什么时候用哪个我在实际项目中观察到很多人有选择困难症其实没那么纠结。如果你做的是数据清洗、ETL、临时报表查询用DataFrame就够了写法简洁SQL风格和DSL风格都能用开发效率高。尤其是处理来自JSON、Parquet、Hive表这些外部数据源时DataFrame天然契合。如果你写的是一个核心业务模块后续代码要长期维护、要单元测试那DataSet更合适。因为case class带来的强类型可以让很多低级错误在编译期就暴露出来。像我在做用户画像模块时就习惯把用户实体定义成case class后续过滤、聚合都在DataSet上做代码可读性好很多。还有一个经验在同一个任务里混着用也没问题DataFrame和DataSet之间可以方便地互转。重要的是理解它们的差异而不是死守着某一种。2. IDEA环境搭建与第一个SparkSession2.1 Maven依赖与本地环境配置工欲善其事必先利其器。Spark项目在IDEA里跑起来其实就是搞定一个Maven工程。我用的是Scala 2.11.12 Spark 2.4.8这个组合非常经典坑也少。Java建议用8虽然新版Spark支持高版本Java但本地调试用老版本最省事。打开IDEA新建一个Maven工程勾选Scala插件支持的骨架然后在pom.xml里加上Spark核心依赖properties scala.version2.11.12/scala.version spark.version2.4.8/spark.version /properties dependencies dependency groupIdorg.scala-lang/groupId artifactIdscala-library/artifactId version${scala.version}/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.11/artifactId version${spark.version}/version /dependency /dependencies注意到这里只加了spark-sql一个依赖就够了因为spark-sql本身会传递依赖spark-core、spark-catalyst这些模块不需要逐一手动添加。这只是最小可用配置。你实际跑的时候如果是在Windows本机调试还需要额外处理Hadoop的winutils问题。我的做法是在项目resources目录下放一个winutils.exe对应的bin目录然后设置环境变量HADOOP_HOME不然提交任务时会报找不到winutils.exe相关的错误。这个坑很多人会踩到后面我单独说。2.2 创建SparkSession时容易被忽略的参数Spark SQL程序的入口是SparkSession它替代了老版本的SparkContext和SQLContext。写Spark SQL代码第一步就是创建并初始化SparkSessionimport org.apache.spark.sql.SparkSession val spark SparkSession .builder() .appName(SparkSQLDataFrameDemo) .master(local[*]) .config(spark.sql.shuffle.partitions, 4) .getOrCreate() val sc spark.sparkContext sc.setLogLevel(WARN)这里每个参数都值得解释一下。local[]表示用本机所有可用的CPU核心来跑星号是通配符。如果只是学习测试也可以写成local[2]固定使用2个核心避免把电脑资源全部吃掉。如果你在集群上跑master通常由提交命令指定不需要硬编码在代码里但本地学习时写上local[]是惯例。spark.sql.shuffle.partitions这个参数很容易被忽略它控制的是Spark SQL执行shuffle操作时的分区数量默认是200。我本地测试果汁量很小如果保持200个分区每次shuffle都会创建一堆空任务白白浪费时间。所以我习惯把它调小到4个让任务调度开销明显降低。setLogLevel(WARN)也是必须的不然控制台会被一堆INFO日志刷屏干扰你看结果。还有一点经验创建完SparkSession之后记得在代码最后调用spark.stop()释放资源。虽然本地程序跑完就退出进程了但在某些长期运行的服务场景里不关Session会导致资源泄露养成好习惯很重要。3. DataFrame的三种创建方式3.1 从JSON文件直接读一等公民的方式学习DataFrame最好的切入点不是自己造数据而是直接读文件因为真实工作里大部分数据都是外部来的。这里我用经典的people.json来演示文件内容很简单{name:Michael, age:29} {name:Andy, age:30} {name:Justin, age:19}在IDEA中把文件放在src/main/resources目录下然后用一行代码就能读成DataFrameval peopleDF spark.read .option(inferSchema, true) .json(src/main/resources/people.json) peopleDF.printSchema() peopleDF.show()输出结果root |-- age: long (nullable true) |-- name: string (nullable true) ---------- |age| name| ---------- | 29|Michael| | 30| Andy| | 19| Justin| ----------注意printSchema()里显示的列顺序是按字母排序的age在前、name在后这和JSON文件里属性出现的顺序无关。这是Spark内部优化导致的属于正常现象不用管它。这里有个细节值得说道option(inferSchema, true)这个参数。它的作用是让Spark自动推断JSON中每一列的数据类型。如果不加这个参数Spark会把所有字段默认推断为string类型。你后面想对age做数值计算的时候就会发现类型对不上还要再cast转换一次很麻烦。我的习惯是所有读外部数据的操作都显式指定inferSchema为true或者干脆使用schema来定义列类型。如果你提前就知道数据结构更推荐显式指定schemaimport org.apache.spark.sql.types._ val schema StructType(Array( StructField(name, StringType, nullable false), StructField(age, IntegerType, nullable true) )) val peopleDF2 spark.read .schema(schema) .json(src/main/resources/people.json)显式指定schema的好处是第一不用做类型推断读取速度更快第二列的类型完全可控不会出现“我以为这是整数结果却是个字符串”的问题第三如果JSON里有多余的脏字段会被直接丢弃。3.2 集合转DataFrametoDF的隐式转换很多时候测试代码不想建文件直接在代码里造点数据用集合转DataFrame是最顺手的。这里会用到Spark的一个隐式转换功能必须先导入spark.implicits._否则编译器找不到toDF方法。import spark.implicits._ val data Seq((Michael, 29), (Andy, 30), (Justin, 19)) val dfFromSeq data.toDF(name, age) dfFromSeq.show()这段代码干的事情很简单Seq里的每个元素是一个二元组toDF传的name, age是列名。Spark会自动把元组变成一行数据。除了元组序列你也可以用case class序列来转这种方式更推荐因为语义更清晰case class Person(name: String, age: Int) val personList Seq(Person(Michael, 29), Person(Andy, 30), Person(Justin, 19)) val personDF personList.toDF()这里连列名都不用指定了case class的字段名自动变成列名。而且列的类型也会按照case class里的类型来推断。toDF这个方式特别适合在测试阶段快速制造DataFrame。我写单元测试时经常用它来造一批假数据再调用真实的数据处理函数验证逻辑是否正确。注意使用toDF之前必须导入spark.implicits._这个导入position很讲究有时候你写代码时idea语法高亮不会报错一运行就报“value toDF is not a member of”多半就是这个隐式转换没有正确导入。3.3 RDD加Schema让普通数据带上结构前面两种方式都算比较高级的API。但在Spark发展早期人们手头很多数据是以RDD形式存在的把RDD转换成DataFrame是一个绕不开的需求。这里有两种典型做法。第一种方式是用反射自动推导case class的字段。假设我们有一个people.txt文件每行格式是“姓名,年龄”Michael,29 Andy,30 Justin,19可以这样读取并转换import spark.implicits._ val peopleRDD sc.textFile(src/main/resources/people.txt) case class People(name: String, age: Int) val peopleDF peopleRDD .map(_.split(,)) .map(p People(p(0).trim, p(1).trim.toInt)) .toDF()这段代码的核心思路是先把RDD里的文本行拆成数组再映射成People对象最后toDF。因为People是case classSpark能自动探测它的字段名和类型不需要额外写schema。第二种方式是手动指定schema适合字段特别多、或者字段结构会变化的场景。做法是先构建一个RDD[Row]再做StructType映射import org.apache.spark.sql.Row import org.apache.spark.sql.types._ val peopleRDD sc.textFile(src/main/resources/people.txt) val schemaString name age val fields schemaString.split( ) .map(fieldName StructField(fieldName, StringType, nullable true)) val schema StructType(fields) val rowRDD peopleRDD .map(_.split(,)) .map(p Row(p(0).trim, p(1).trim)) val peopleDF spark.createDataFrame(rowRDD, schema)这里所有列的类型都是String使用时需要判断是否需要cast。手动指定schema的灵活性更高适用于列很多、且来自配置文件或元数据服务的情况。对比两种方式我的习惯是能用case class反射就用反射因为代码量少并且类型安全。只有当字段数量多到case class维护困难或者数据字段经常动态变化时才选择手动schema。4. DataSet的创建与类型安全的操作4.1 用case class划出DataSet的边界DataSet和DataFrame最大的区别就是强类型。所以创建一个DataSet关键步骤是定义一个case class作为数据行的类型。沿用前面的People例子import spark.implicits._ case class Person(name: String, age: Int) val personDS Seq(Person(Michael, 29), Person(Andy, 30), Person(Justin, 19)).toDS() personDS.show()这里调用的toDS()和之前用的toDF()类似也是spark.implicits提供的隐式转换。区别在于toDS返回的是DataSet[Person]每一行都是一个Person对象而不是一个Row。你还可以从JSON文件直接读到DataSetcase class Person(name: String, age: Int) val personDS spark.read .option(inferSchema, true) .json(src/main/resources/people.json) .as[Person]关键在这一句.as[Person]它把DataFrame转换成DataSet[Person]。注意这个转换要求DataFrame的列名和类型与case class的字段名和类型对得上。名字不一致的话要么报错要么字段全是null这一点我后面避坑部分会细讲。4.2 类型安全API和lambda表达式配合的写法DataSet真正的强项在于它是可以用函数式风格操作的操作的都是真实的类对象不再有Row类型那种“拿出一个字段还要asInstanceOf”的麻烦。比如我们要过滤出年龄大于等于18岁的人val adultsDS personDS.filter(p p.age 18) adultsDS.map(p s${p.name} is ${p.age} years old).show(false)在filter里面p就是Person类型写p.age编译器能直接帮你校验字段是否存在。这个和DataFrame的写法差异很大在DataFrame上写filter参数需要操作Column用的是$age这种语法编译期不会检查字段名。再比如做聚合统计case class AgeCount(age: Int, cnt: Long) val ageCountDS personDS .groupBy(age) .count() .as[AgeCount] .map(ac sage ${ac.age} appears ${ac.count})groupBy(age)返回的是一个“分组后的DataFrame”count()返回的是DataFrame[Row]最后再用.as[AgeCount]转换成强类型的DataSet。后面map里ac.age就能直接访问。这种写法在管道比较长时可读性优势非常明显。从实践角度看DataSet的强类型特性很适合做业务逻辑封装。把数据过滤、计算规则封装成一个个函数入参出参都用case class跑单元测试的时候特别舒服。如果你的团队里约定核心计算逻辑都用DataSet 自定义函数后期维护成本会低很多。4.3 DataFrame与DataSet相互转换细节实际开发里DataFrame和DataSet经常要互相转换。DataFrame转DataSet是as加类型val personDS peopleDF.as[Person]反向转换更简单DataSet天然就是DataFrame的子接口所以直接当成DataFrame用就行或者调用toDF方法变成普通DataFrameval personDF personDS.toDF()这里想提醒你一个坑DataSet转DataFrame之后行类型变成了Row原来case class里字段的类型如果和预期的Schema不一致后续操作可能出现类型不匹配。比如Person.age是Int转成DataFrame后某些函数会按Long处理当你再as回别的case class时就会报错。应对的办法是在相互转换的节点用printSchema()打印一下结构确认类型没有突变。强烈建议每个转换的代码段后面都接一个printSchema()视觉上确认schema是否符合预期。补充一个经验DataFrame的列默认允许null但case class的字段并没有nullability概念所以从DataFrame转DataSet时如果某一列有null值可能出现“AssertionError: null”之类的运行时错误。用nullable false定义schema无法完全杜绝最好在转换前对数据做空值处理比如df.na.fill()、df.na.fill(0)或者过滤掉null。5. 常见报错与排查实录这部分我把自己之前趟过的坑和群里同学问得比较多的问题整理成一张表报错信息原因解决办法value toDF is not a member of Seq没有导入spark.implicits._在代码里加import spark.implicits._Unable to find encoder for type stored in a Dataset使用的case class定义在了函数内部把case class定义移到main方法外部或对象顶层Exception: null / AssertionError: nullDataFrame的列存在null无法转换为强类型字段转换前用na.drop()或na.fill()处理空值mismatched input . expecting (DSL语法写错比如情绪化调用检查操作符优先级必要时加括号Job aborted due to stage failure: Task not serializable在算子中使用了非序列化的外部对象把引用变量改成局部变量或者用广播变量No applicable method for encodeRef类型不是case class换成case class不要用普通class除了表格里这些我再展开讲两个高频场景。第一个是case class位置问题。很多人写Scala容易把case class定义在main方法内部这在Spark里经常会报encoder找不到。因为Spark需要为case class生成隐式编码器它要求case class必须定义在顶层或者object下。我在IDEA里遇到这个报错时第一反应就是把case class挪到外层。还有一个非常容易踩的坑是case class字段名和JSON里的字段名不一致。比如JSON里是user_namecase class里是userName这在as[Person]转换时不会报错但对应字段会被填成null。这个bug很难排查因为代码不会中断只是数据莫名其妙变成null。我的排查方法是先show()看原始DataFrame再as后show()对比马上就能看出哪列丢了。第三个高频问题是Windows下运行时找不到winutils.exe。表现是运行任何Spark程序控制台先输出一段Hadoop native相关的WARN有些严格场景会直接报错退出。解决办法是下载对应版本的winutils放到一个本地目录把目录配成HADOOP_HOME系统环境变量。虽然Spark SQL本地跑很多功能不依赖HDFS但Spark初始化时还是会去检查Hadoop本地库。再来说一个关于StringIndexer和类型推断的经典场景。你从CSV读入一个age字段以为是数值类型结果show()出来的全是字符串。原因多半是没有设置inferSchema为true而CSV默认所有列都是string。你后面做pivot、数值计算时到处出问题根源就在这里。前置地printSchema()一眼就能看出来列类型显示string就说明推断没生效。6. 实操心得与一些建议代码写到这里DataFrame和DataSet的基础用法你已经能跑通了。最后分享几个我在实际项目里沉淀下来的经验和口味算是我自己的偏好。首先在Spark SQL里SQL风格的写法依然值得学。DataFrame支持注册成临时视图然后直接用sql()方法写标准SQL。比如刚才的peopleDF我可以这么用peopleDF.createOrReplaceTempView(people) val result spark.sql( |SELECT name, age 1 AS age_next_year |FROM people |WHERE age 18 |.stripMargin) result.show()这种方式在处理复杂报表需求时极其方便尤其适合团队里比较熟悉SQL但不熟悉函数式API的成员。我在实际项目中经常是“两碗水端平”简单过滤、映射用DSL复杂多表关联、子查询用SQL这样代码可读性更高。其次不论用DataFrame还是DataSet都要警惕“小文件”问题。如果你在本地测试时把之前说的shuffle分区数调成4确实快了但如果在集群上用默认200分区跑了一个很小的作业每个分区写入一个文件就会产生200个小文件后续再读这些文件时性能非常差。我在产出结果表时通常会在写文件前用repartition(1)或coalesce(1)控制分区数或者用sortByPartition保证后续读取效率。最后从DataFrame转DataSet时我一直建议在关键节点做一次describe()或show()不要信“代码能编译就说明结果正确”。Spark SQL开发里数据质量问题远比代码逻辑问题多空值、类型偏差、列名漂移都可能在不经意间污染你的结果。加上printSchema和sample数据的检查能提前拦截80%以上的问题。Spark SQL本身迭代很快现在很多人在用3.x甚至更新的版本但本文这套DataFrame和DataSet核心思路是稳定不变的。熟练了基础API之后后面学Structured Streaming、Catalyst优化、Spark调优都会更有底气。建议你打开IDEA照着代码敲一遍再试着改一改过滤条件、换一换数据类型把这篇的每个例子都跑通再去找更多复杂案例结合着练。DataFrame和DataSet这两个API在工业界的地位非常稳固值得花时间吃透。 
长河君 · 2026/10/13 7:46:34 · 阅读 1.2k Spark SQL入门实战：DataFrame与DataSet创建、转换及避坑指南
我记得刚开始接触大数据的时候最绕不开的一个坎就是Spark。而Spark里面入门门槛最低、后面工作里用得最多的就是Spark SQL的DataFrame和DataSet。很多教程要么直接贴官网文档要么甩一堆概念字符串、列、Schema、编码器……看完一头雾水开IDEA还是不知道从哪下手。这篇文章我就按我自己上手时候的顺序来直接用IDEA搭一个Maven工程从SparkSession初始化开始把DataFrame和DataSet的创建、转换、常见坑全部过一遍全程代码可复制。适合刚学完Scala基础、准备进坑Spark的人也适合用Java写Spark但想对比一下两种API差异的朋友。1. 动手前的整体思考为什么DataFrame与DataSet值得专门学1.1 DataFrame与DataSet的身世从RDD到更友好的抽象很多教程喜欢从RDD开始讲导致很多人以为RDD是Spark的“本体”。实际上RDD确实是Spark最初的抽象但它有两个很明显的痛点一是没有schema结构数据是什么字段、什么类型全靠写代码的人自己记二是性能优化全靠程序员手动做了能做的优化也有限。后来Spark SQL出现了带来了DataFrame。DataFrame是什么你可以把DataFrame看成一张分布式的表有列名、有列类型像数据库表一样。它的底层仍然是RDD但外表换了一套更友好的API。更重要的是Spark对DataFrame做了Catalyst优化器简单说就是Spark会自动帮你想“这个查询怎么写更快”很多性能优化在内部就完成了不需要你操心。DataSet呢它是在DataFrame基础上又加了一层类型约束。DataFrame的每一行是Row类型这种是弱类型的列的类型错了要到运行期才发现。而DataSet每一行是一个具体的case class对象比如People、Order这些编译期就能检查出类型对不对。换句话说DataFrame DataSet[Row]这是理解两者的一个关键公式。1.2 两个API各自的脾气什么时候用哪个我在实际项目中观察到很多人有选择困难症其实没那么纠结。如果你做的是数据清洗、ETL、临时报表查询用DataFrame就够了写法简洁SQL风格和DSL风格都能用开发效率高。尤其是处理来自JSON、Parquet、Hive表这些外部数据源时DataFrame天然契合。如果你写的是一个核心业务模块后续代码要长期维护、要单元测试那DataSet更合适。因为case class带来的强类型可以让很多低级错误在编译期就暴露出来。像我在做用户画像模块时就习惯把用户实体定义成case class后续过滤、聚合都在DataSet上做代码可读性好很多。还有一个经验在同一个任务里混着用也没问题DataFrame和DataSet之间可以方便地互转。重要的是理解它们的差异而不是死守着某一种。2. IDEA环境搭建与第一个SparkSession2.1 Maven依赖与本地环境配置工欲善其事必先利其器。Spark项目在IDEA里跑起来其实就是搞定一个Maven工程。我用的是Scala 2.11.12 Spark 2.4.8这个组合非常经典坑也少。Java建议用8虽然新版Spark支持高版本Java但本地调试用老版本最省事。打开IDEA新建一个Maven工程勾选Scala插件支持的骨架然后在pom.xml里加上Spark核心依赖properties scala.version2.11.12/scala.version spark.version2.4.8/spark.version /properties dependencies dependency groupIdorg.scala-lang/groupId artifactIdscala-library/artifactId version${scala.version}/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.11/artifactId version${spark.version}/version /dependency /dependencies注意到这里只加了spark-sql一个依赖就够了因为spark-sql本身会传递依赖spark-core、spark-catalyst这些模块不需要逐一手动添加。这只是最小可用配置。你实际跑的时候如果是在Windows本机调试还需要额外处理Hadoop的winutils问题。我的做法是在项目resources目录下放一个winutils.exe对应的bin目录然后设置环境变量HADOOP_HOME不然提交任务时会报找不到winutils.exe相关的错误。这个坑很多人会踩到后面我单独说。2.2 创建SparkSession时容易被忽略的参数Spark SQL程序的入口是SparkSession它替代了老版本的SparkContext和SQLContext。写Spark SQL代码第一步就是创建并初始化SparkSessionimport org.apache.spark.sql.SparkSession val spark SparkSession .builder() .appName(SparkSQLDataFrameDemo) .master(local[*]) .config(spark.sql.shuffle.partitions, 4) .getOrCreate() val sc spark.sparkContext sc.setLogLevel(WARN)这里每个参数都值得解释一下。local[]表示用本机所有可用的CPU核心来跑星号是通配符。如果只是学习测试也可以写成local[2]固定使用2个核心避免把电脑资源全部吃掉。如果你在集群上跑master通常由提交命令指定不需要硬编码在代码里但本地学习时写上local[]是惯例。spark.sql.shuffle.partitions这个参数很容易被忽略它控制的是Spark SQL执行shuffle操作时的分区数量默认是200。我本地测试果汁量很小如果保持200个分区每次shuffle都会创建一堆空任务白白浪费时间。所以我习惯把它调小到4个让任务调度开销明显降低。setLogLevel(WARN)也是必须的不然控制台会被一堆INFO日志刷屏干扰你看结果。还有一点经验创建完SparkSession之后记得在代码最后调用spark.stop()释放资源。虽然本地程序跑完就退出进程了但在某些长期运行的服务场景里不关Session会导致资源泄露养成好习惯很重要。3. DataFrame的三种创建方式3.1 从JSON文件直接读一等公民的方式学习DataFrame最好的切入点不是自己造数据而是直接读文件因为真实工作里大部分数据都是外部来的。这里我用经典的people.json来演示文件内容很简单{name:Michael, age:29} {name:Andy, age:30} {name:Justin, age:19}在IDEA中把文件放在src/main/resources目录下然后用一行代码就能读成DataFrameval peopleDF spark.read .option(inferSchema, true) .json(src/main/resources/people.json) peopleDF.printSchema() peopleDF.show()输出结果root |-- age: long (nullable true) |-- name: string (nullable true) ---------- |age| name| ---------- | 29|Michael| | 30| Andy| | 19| Justin| ----------注意printSchema()里显示的列顺序是按字母排序的age在前、name在后这和JSON文件里属性出现的顺序无关。这是Spark内部优化导致的属于正常现象不用管它。这里有个细节值得说道option(inferSchema, true)这个参数。它的作用是让Spark自动推断JSON中每一列的数据类型。如果不加这个参数Spark会把所有字段默认推断为string类型。你后面想对age做数值计算的时候就会发现类型对不上还要再cast转换一次很麻烦。我的习惯是所有读外部数据的操作都显式指定inferSchema为true或者干脆使用schema来定义列类型。如果你提前就知道数据结构更推荐显式指定schemaimport org.apache.spark.sql.types._ val schema StructType(Array( StructField(name, StringType, nullable false), StructField(age, IntegerType, nullable true) )) val peopleDF2 spark.read .schema(schema) .json(src/main/resources/people.json)显式指定schema的好处是第一不用做类型推断读取速度更快第二列的类型完全可控不会出现“我以为这是整数结果却是个字符串”的问题第三如果JSON里有多余的脏字段会被直接丢弃。3.2 集合转DataFrametoDF的隐式转换很多时候测试代码不想建文件直接在代码里造点数据用集合转DataFrame是最顺手的。这里会用到Spark的一个隐式转换功能必须先导入spark.implicits._否则编译器找不到toDF方法。import spark.implicits._ val data Seq((Michael, 29), (Andy, 30), (Justin, 19)) val dfFromSeq data.toDF(name, age) dfFromSeq.show()这段代码干的事情很简单Seq里的每个元素是一个二元组toDF传的name, age是列名。Spark会自动把元组变成一行数据。除了元组序列你也可以用case class序列来转这种方式更推荐因为语义更清晰case class Person(name: String, age: Int) val personList Seq(Person(Michael, 29), Person(Andy, 30), Person(Justin, 19)) val personDF personList.toDF()这里连列名都不用指定了case class的字段名自动变成列名。而且列的类型也会按照case class里的类型来推断。toDF这个方式特别适合在测试阶段快速制造DataFrame。我写单元测试时经常用它来造一批假数据再调用真实的数据处理函数验证逻辑是否正确。注意使用toDF之前必须导入spark.implicits._这个导入position很讲究有时候你写代码时idea语法高亮不会报错一运行就报“value toDF is not a member of”多半就是这个隐式转换没有正确导入。3.3 RDD加Schema让普通数据带上结构前面两种方式都算比较高级的API。但在Spark发展早期人们手头很多数据是以RDD形式存在的把RDD转换成DataFrame是一个绕不开的需求。这里有两种典型做法。第一种方式是用反射自动推导case class的字段。假设我们有一个people.txt文件每行格式是“姓名,年龄”Michael,29 Andy,30 Justin,19可以这样读取并转换import spark.implicits._ val peopleRDD sc.textFile(src/main/resources/people.txt) case class People(name: String, age: Int) val peopleDF peopleRDD .map(_.split(,)) .map(p People(p(0).trim, p(1).trim.toInt)) .toDF()这段代码的核心思路是先把RDD里的文本行拆成数组再映射成People对象最后toDF。因为People是case classSpark能自动探测它的字段名和类型不需要额外写schema。第二种方式是手动指定schema适合字段特别多、或者字段结构会变化的场景。做法是先构建一个RDD[Row]再做StructType映射import org.apache.spark.sql.Row import org.apache.spark.sql.types._ val peopleRDD sc.textFile(src/main/resources/people.txt) val schemaString name age val fields schemaString.split( ) .map(fieldName StructField(fieldName, StringType, nullable true)) val schema StructType(fields) val rowRDD peopleRDD .map(_.split(,)) .map(p Row(p(0).trim, p(1).trim)) val peopleDF spark.createDataFrame(rowRDD, schema)这里所有列的类型都是String使用时需要判断是否需要cast。手动指定schema的灵活性更高适用于列很多、且来自配置文件或元数据服务的情况。对比两种方式我的习惯是能用case class反射就用反射因为代码量少并且类型安全。只有当字段数量多到case class维护困难或者数据字段经常动态变化时才选择手动schema。4. DataSet的创建与类型安全的操作4.1 用case class划出DataSet的边界DataSet和DataFrame最大的区别就是强类型。所以创建一个DataSet关键步骤是定义一个case class作为数据行的类型。沿用前面的People例子import spark.implicits._ case class Person(name: String, age: Int) val personDS Seq(Person(Michael, 29), Person(Andy, 30), Person(Justin, 19)).toDS() personDS.show()这里调用的toDS()和之前用的toDF()类似也是spark.implicits提供的隐式转换。区别在于toDS返回的是DataSet[Person]每一行都是一个Person对象而不是一个Row。你还可以从JSON文件直接读到DataSetcase class Person(name: String, age: Int) val personDS spark.read .option(inferSchema, true) .json(src/main/resources/people.json) .as[Person]关键在这一句.as[Person]它把DataFrame转换成DataSet[Person]。注意这个转换要求DataFrame的列名和类型与case class的字段名和类型对得上。名字不一致的话要么报错要么字段全是null这一点我后面避坑部分会细讲。4.2 类型安全API和lambda表达式配合的写法DataSet真正的强项在于它是可以用函数式风格操作的操作的都是真实的类对象不再有Row类型那种“拿出一个字段还要asInstanceOf”的麻烦。比如我们要过滤出年龄大于等于18岁的人val adultsDS personDS.filter(p p.age 18) adultsDS.map(p s${p.name} is ${p.age} years old).show(false)在filter里面p就是Person类型写p.age编译器能直接帮你校验字段是否存在。这个和DataFrame的写法差异很大在DataFrame上写filter参数需要操作Column用的是$age这种语法编译期不会检查字段名。再比如做聚合统计case class AgeCount(age: Int, cnt: Long) val ageCountDS personDS .groupBy(age) .count() .as[AgeCount] .map(ac sage ${ac.age} appears ${ac.count})groupBy(age)返回的是一个“分组后的DataFrame”count()返回的是DataFrame[Row]最后再用.as[AgeCount]转换成强类型的DataSet。后面map里ac.age就能直接访问。这种写法在管道比较长时可读性优势非常明显。从实践角度看DataSet的强类型特性很适合做业务逻辑封装。把数据过滤、计算规则封装成一个个函数入参出参都用case class跑单元测试的时候特别舒服。如果你的团队里约定核心计算逻辑都用DataSet 自定义函数后期维护成本会低很多。4.3 DataFrame与DataSet相互转换细节实际开发里DataFrame和DataSet经常要互相转换。DataFrame转DataSet是as加类型val personDS peopleDF.as[Person]反向转换更简单DataSet天然就是DataFrame的子接口所以直接当成DataFrame用就行或者调用toDF方法变成普通DataFrameval personDF personDS.toDF()这里想提醒你一个坑DataSet转DataFrame之后行类型变成了Row原来case class里字段的类型如果和预期的Schema不一致后续操作可能出现类型不匹配。比如Person.age是Int转成DataFrame后某些函数会按Long处理当你再as回别的case class时就会报错。应对的办法是在相互转换的节点用printSchema()打印一下结构确认类型没有突变。强烈建议每个转换的代码段后面都接一个printSchema()视觉上确认schema是否符合预期。补充一个经验DataFrame的列默认允许null但case class的字段并没有nullability概念所以从DataFrame转DataSet时如果某一列有null值可能出现“AssertionError: null”之类的运行时错误。用nullable false定义schema无法完全杜绝最好在转换前对数据做空值处理比如df.na.fill()、df.na.fill(0)或者过滤掉null。5. 常见报错与排查实录这部分我把自己之前趟过的坑和群里同学问得比较多的问题整理成一张表报错信息原因解决办法value toDF is not a member of Seq没有导入spark.implicits._在代码里加import spark.implicits._Unable to find encoder for type stored in a Dataset使用的case class定义在了函数内部把case class定义移到main方法外部或对象顶层Exception: null / AssertionError: nullDataFrame的列存在null无法转换为强类型字段转换前用na.drop()或na.fill()处理空值mismatched input . expecting (DSL语法写错比如情绪化调用检查操作符优先级必要时加括号Job aborted due to stage failure: Task not serializable在算子中使用了非序列化的外部对象把引用变量改成局部变量或者用广播变量No applicable method for encodeRef类型不是case class换成case class不要用普通class除了表格里这些我再展开讲两个高频场景。第一个是case class位置问题。很多人写Scala容易把case class定义在main方法内部这在Spark里经常会报encoder找不到。因为Spark需要为case class生成隐式编码器它要求case class必须定义在顶层或者object下。我在IDEA里遇到这个报错时第一反应就是把case class挪到外层。还有一个非常容易踩的坑是case class字段名和JSON里的字段名不一致。比如JSON里是user_namecase class里是userName这在as[Person]转换时不会报错但对应字段会被填成null。这个bug很难排查因为代码不会中断只是数据莫名其妙变成null。我的排查方法是先show()看原始DataFrame再as后show()对比马上就能看出哪列丢了。第三个高频问题是Windows下运行时找不到winutils.exe。表现是运行任何Spark程序控制台先输出一段Hadoop native相关的WARN有些严格场景会直接报错退出。解决办法是下载对应版本的winutils放到一个本地目录把目录配成HADOOP_HOME系统环境变量。虽然Spark SQL本地跑很多功能不依赖HDFS但Spark初始化时还是会去检查Hadoop本地库。再来说一个关于StringIndexer和类型推断的经典场景。你从CSV读入一个age字段以为是数值类型结果show()出来的全是字符串。原因多半是没有设置inferSchema为true而CSV默认所有列都是string。你后面做pivot、数值计算时到处出问题根源就在这里。前置地printSchema()一眼就能看出来列类型显示string就说明推断没生效。6. 实操心得与一些建议代码写到这里DataFrame和DataSet的基础用法你已经能跑通了。最后分享几个我在实际项目里沉淀下来的经验和口味算是我自己的偏好。首先在Spark SQL里SQL风格的写法依然值得学。DataFrame支持注册成临时视图然后直接用sql()方法写标准SQL。比如刚才的peopleDF我可以这么用peopleDF.createOrReplaceTempView(people) val result spark.sql( |SELECT name, age 1 AS age_next_year |FROM people |WHERE age 18 |.stripMargin) result.show()这种方式在处理复杂报表需求时极其方便尤其适合团队里比较熟悉SQL但不熟悉函数式API的成员。我在实际项目中经常是“两碗水端平”简单过滤、映射用DSL复杂多表关联、子查询用SQL这样代码可读性更高。其次不论用DataFrame还是DataSet都要警惕“小文件”问题。如果你在本地测试时把之前说的shuffle分区数调成4确实快了但如果在集群上用默认200分区跑了一个很小的作业每个分区写入一个文件就会产生200个小文件后续再读这些文件时性能非常差。我在产出结果表时通常会在写文件前用repartition(1)或coalesce(1)控制分区数或者用sortByPartition保证后续读取效率。最后从DataFrame转DataSet时我一直建议在关键节点做一次describe()或show()不要信“代码能编译就说明结果正确”。Spark SQL开发里数据质量问题远比代码逻辑问题多空值、类型偏差、列名漂移都可能在不经意间污染你的结果。加上printSchema和sample数据的检查能提前拦截80%以上的问题。Spark SQL本身迭代很快现在很多人在用3.x甚至更新的版本但本文这套DataFrame和DataSet核心思路是稳定不变的。熟练了基础API之后后面学Structured Streaming、Catalyst优化、Spark调优都会更有底气。建议你打开IDEA照着代码敲一遍再试着改一改过滤条件、换一换数据类型把这篇的每个例子都跑通再去找更多复杂案例结合着练。DataFrame和DataSet这两个API在工业界的地位非常稳固值得花时间吃透。