服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - Java和scala实现 Spark RDD转换成DataFrame的两种方法小结

Java和scala实现 Spark RDD转换成DataFrame的两种方法小结

2021-05-07 13:34黑白调92 Java教程

今天小编就为大家分享一篇Java和scala实现 Spark RDD转换成DataFrame的两种方法小结,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

一:准备数据源

在项目下新建一个student.txt文件,里面的内容为:

?
1
2
3
4
1,zhangsan,20
2,lisi,21
3,wanger,19
4,fangliu,18

二:实现

java版:

1.首先新建一个student的bean对象,实现序列化和tostring()方法,具体代码如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com.cxd.sql;
import java.io.serializable;
@suppresswarnings("serial")
public class student implements serializable {
 string sid;
 string sname;
 int sage;
 public string getsid() {
  return sid;
 }
 public void setsid(string sid) {
  this.sid = sid;
 }
 public string getsname() {
  return sname;
 }
 public void setsname(string sname) {
  this.sname = sname;
 }
 public int getsage() {
  return sage;
 }
 public void setsage(int sage) {
  this.sage = sage;
 }
 @override
 public string tostring() {
  return "student [sid=" + sid + ", sname=" + sname + ", sage=" + sage + "]";
 }
 
}
       

2.转换,具体代码如下

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package com.cxd.sql;
import java.util.arraylist;
import org.apache.spark.sparkconf;
import org.apache.spark.api.java.javardd;
import org.apache.spark.sql.dataset;
import org.apache.spark.sql.row;
import org.apache.spark.sql.rowfactory;
import org.apache.spark.sql.savemode;
import org.apache.spark.sql.sparksession;
import org.apache.spark.sql.types.datatypes;
import org.apache.spark.sql.types.structfield;
import org.apache.spark.sql.types.structtype;
public class txttoparquetdemo {
 public static void main(string[] args) {
  
  sparkconf conf = new sparkconf().setappname("txttoparquet").setmaster("local");
  sparksession spark = sparksession.builder().config(conf).getorcreate();
  reflecttransform(spark);//java反射
  dynamictransform(spark);//动态转换
 }
 
 /**
  * 通过java反射转换
  * @param spark
  */
 private static void reflecttransform(sparksession spark)
 {
  javardd<string> source = spark.read().textfile("stuinfo.txt").javardd();
  
  javardd<student> rowrdd = source.map(line -> {
   string parts[] = line.split(",");
   student stu = new student();
   stu.setsid(parts[0]);
   stu.setsname(parts[1]);
   stu.setsage(integer.valueof(parts[2]));
   return stu;
  });
  
  dataset<row> df = spark.createdataframe(rowrdd, student.class);
  df.select("sid", "sname", "sage").
  coalesce(1).write().mode(savemode.append).parquet("parquet.res");
 }
 /**
  * 动态转换
  * @param spark
  */
 private static void dynamictransform(sparksession spark)
 {
  javardd<string> source = spark.read().textfile("stuinfo.txt").javardd();
  
  javardd<row> rowrdd = source.map( line -> {
   string[] parts = line.split(",");
   string sid = parts[0];
   string sname = parts[1];
   int sage = integer.parseint(parts[2]);
   
   return rowfactory.create(
     sid,
     sname,
     sage
     );
  });
  
  arraylist<structfield> fields = new arraylist<structfield>();
  structfield field = null;
  field = datatypes.createstructfield("sid", datatypes.stringtype, true);
  fields.add(field);
  field = datatypes.createstructfield("sname", datatypes.stringtype, true);
  fields.add(field);
  field = datatypes.createstructfield("sage", datatypes.integertype, true);
  fields.add(field);
  
  structtype schema = datatypes.createstructtype(fields);
  
  dataset<row> df = spark.createdataframe(rowrdd, schema);
  df.coalesce(1).write().mode(savemode.append).parquet("parquet.res1");
  
  
 }
 
}

scala版本:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import org.apache.spark.sql.sparksession
import org.apache.spark.sql.types.stringtype
import org.apache.spark.sql.types.structfield
import org.apache.spark.sql.types.structtype
import org.apache.spark.sql.row
import org.apache.spark.sql.types.integertype
object rdd2dataset {
 
 case class student(id:int,name:string,age:int)
 def main(args:array[string])
 {
 
 val spark=sparksession.builder().master("local").appname("rdd2dataset").getorcreate()
 import spark.implicits._
 reflectcreate(spark)
 dynamiccreate(spark)
 }
 
 /**
     * 通过java反射转换
     * @param spark
     */
 private def reflectcreate(spark:sparksession):unit={
 import spark.implicits._
 val sturdd=spark.sparkcontext.textfile("student2.txt")
 //todf()为隐式转换
 val studf=sturdd.map(_.split(",")).map(parts⇒student(parts(0).trim.toint,parts(1),parts(2).trim.toint)).todf()
 //studf.select("id","name","age").write.text("result") //对写入文件指定列名
 studf.printschema()
 studf.createorreplacetempview("student")
 val namedf=spark.sql("select name from student where age<20")
 //namedf.write.text("result") //将查询结果写入一个文件
 namedf.show()
 }
 
 /**
     * 动态转换
     * @param spark
     */
 private def dynamiccreate(spark:sparksession):unit={
 val sturdd=spark.sparkcontext.textfile("student.txt")
 import spark.implicits._
 val schemastring="id,name,age"
 val fields=schemastring.split(",").map(fieldname => structfield(fieldname, stringtype, nullable = true))
 val schema=structtype(fields)
 val rowrdd=sturdd.map(_.split(",")).map(parts⇒row(parts(0),parts(1),parts(2)))
 val studf=spark.createdataframe(rowrdd, schema)
  studf.printschema()
 val tmpview=studf.createorreplacetempview("student")
 val namedf=spark.sql("select name from student where age<20")
 //namedf.write.text("result") //将查询结果写入一个文件
 namedf.show()
 }
}

注:

1.上面代码全都已经测试通过,测试的环境为spark2.1.0,jdk1.8。

2.此代码不适用于spark2.0以前的版本。

以上这篇java和scala实现 spark rdd转换成dataframe的两种方法小结就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持服务器之家。

原文链接:https://blog.csdn.net/u010592112/article/details/73730796

延伸 · 阅读

精彩推荐