Spark课后练习来了

想要筑建高楼，必先打好地基。学好基础才是走向大牛的关键。废话不多说，各位，请接题。

不好意思，拿错了。下边的才是 ↓ ↓ ↓

文章目录

Spark课后练习来了
- 1、创建一个1-10数组的RDD，将所有元素*2形成新的RDD
- 2、创建一个10-20数组的RDD，使用mapPartitions将所有元素*2形成新的RDD
- 3、创建一个元素为 1-5 的RDD，运用 flatMap创建一个新的 RDD，新的 RDD 为原 RDD 每个元素的平方和三次方来组成 1,1,4,8,9,27.. 4、创建一个 4 个分区的 RDD数据为Array(10,20,30,40,50,60)，使用glom将每个分区的数据放到一个数组
- 4、创建一个 4 个分区的 RDD数据为Array(10,20,30,40,50,60)，使用glom将每个分区的数据放到一个数组
- 5、创建一个 RDD数据为Array(1, 3, 4, 20, 4, 5, 8)，按照元素的奇偶性进行分组
- 6、创建一个 RDD（由字符串组成）Array("xiaoli", "laoli", "laowang", "xiaocang", "xiaojing", "xiaokong")，过滤出一个新 RDD（包含“xiao”子串）
- 7、创建一个 RDD数据为1 to 10，请使用sample不放回抽样
- 8、创建一个 RDD数据为1 to 10，请使用sample放回抽样
- 9、创建一个 RDD数据为Array(10,10,2,5,3,5,3,6,9,1),对 RDD 中元素执行去重操作
- 10、创建一个分区数为5的 RDD，数据为0 to 100，之后使用coalesce再重新减少分区的数量至 2
- 11、创建一个分区数为5的 RDD，数据为0 to 100，之后使用repartition再重新减少分区的数量至 3
- 12、创建一个 RDD数据为1,3,4,10,4,6,9,20,30,16,请给RDD进行分别进行升序和降序排列
- 13、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，求并集
- 14、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算差集，两个都算
- 15、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算交集
- 16、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算 2 个 RDD 的笛卡尔积
- 17、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 5和11 to 15，对两个RDD拉链操作
- 18、创建一个RDD数据为List(("female",1),("male",5),("female",5),("male",2))，请计算出female和male的总数分别为多少
- 19、创建一个有两个分区的 RDD数据为List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8))，取出每个分区相同key对应值的最大值，然后相加
- 20、创建一个有两个分区的 pairRDD数据为Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))，根据 key 计算每种 key 的value的平均值
- 21、统计出每一个省份广告被点击次数的 TOP3，数据在access.log文件中
- 22、读取本地文件words.txt,统计出每个单词的个数，保存数据到 hdfs 上
- 23、读取 people.json 数据的文件, 每行是一个 json 对象，进行解析输出
- 24、保存一个 SequenceFile 文件，使用spark创建一个RDD数据为Array(("a", 1),("b", 2),("c", 3))，保存为SequenceFile格式的文件到hdfs上
- 25、读取24题的SequenceFile 文件并输出
- 26、读写 objectFile 文件，把 RDD 保存为objectFile，RDD数据为Array(("a", 1),("b", 2),("c", 3))，并进行读取出来
- 27、使用内置累加器计算Accumulator.txt文件中空行的数量
- 28、使用Spark广播变量
- 29、mysql创建一个数据库bigdata0407，在此数据库中创建一张表
- 30、在hbase中创建一个表student，有一个 message列族

1、创建一个1-10数组的RDD，将所有元素*2形成新的RDD

val rdd: RDD[Int] = sc.makeRDD(1 to 10)
val resultOne: RDD[Int] = rdd.map(_ * 2)
println(resultOne.collect().toList)

2、创建一个10-20数组的RDD，使用mapPartitions将所有元素*2形成新的RDD

    val rdd2: RDD[Int] = sc.makeRDD(Array(10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20))val resultTwo: RDD[Int] = rdd2.mapPartitions(x => x.map(_ * 2))println(resultTwo.collect().toList)

3、创建一个元素为 1-5 的RDD，运用 flatMap创建一个新的 RDD，新的 RDD 为原 RDD 每个元素的平方和三次方来组成 1,1,4,8,9,27… 4、创建一个 4 个分区的 RDD数据为Array(10,20,30,40,50,60)，使用glom将每个分区的数据放到一个数组

 val rdd3: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))val resultThree: RDD[Int] = rdd3.flatMap(x => Array(x * x, x * x * x))println(resultThree.collect().toList)

4、创建一个 4 个分区的 RDD数据为Array(10,20,30,40,50,60)，使用glom将每个分区的数据放到一个数组

    val rdd4: RDD[Int] = sc.makeRDD(Array(10, 20, 30, 40, 50, 60),4)val resultFour: RDD[Array[Int]] = rdd4.glom()resultFour.foreach(x => println(x.toList))

5、创建一个 RDD数据为Array(1, 3, 4, 20, 4, 5, 8)，按照元素的奇偶性进行分组

 val rdd5: RDD[Int] = sc.makeRDD(Array(1, 3, 4, 20, 4, 5, 8))val tuples: Array[(String, Iterable[Int])] = rdd5.groupBy(x => if (x % 2 == 0) "even" else "odd").collect()println(tuples.toList)

6、创建一个 RDD（由字符串组成）Array(“xiaoli”, “laoli”, “laowang”, “xiaocang”, “xiaojing”, “xiaokong”)，过滤出一个新 RDD（包含“xiao”子串）

 val rdd6: RDD[String] = sc.makeRDD(Array("xiaoli", "laoli", "laowang", "xiaocang", "xiaojing", "xiaokong"))val resultSix: List[String] = rdd6.filter(_.contains("xiao")).collect().toListprintln(resultSix)

7、创建一个 RDD数据为1 to 10，请使用sample不放回抽样

    val rdd7: RDD[Int] = sc.makeRDD(1 to 10)val resultSeven: Array[Int] = rdd7.sample(false, 0.5).collect()println(resultSeven.toList)

8、创建一个 RDD数据为1 to 10，请使用sample放回抽样

   val resultEight: Array[Int] = rdd7.sample(true, 0.5).collect()println(resultEight.toList)

9、创建一个 RDD数据为Array(10,10,2,5,3,5,3,6,9,1),对 RDD 中元素执行去重操作

   val rdd9: RDD[Int] = sc.makeRDD(Array(10, 10, 2, 5, 3, 5, 3, 6, 9, 1))val resultnine: List[Int] = rdd9.distinct().collect().toList
//    println(resultnine)

10、创建一个分区数为5的 RDD，数据为0 to 100，之后使用coalesce再重新减少分区的数量至 2

   val rdd10: RDD[Int] = sc.makeRDD(0 to 100, 5)val resultTen: RDD[Int] = rdd10.coalesce(2)

11、创建一个分区数为5的 RDD，数据为0 to 100，之后使用repartition再重新减少分区的数量至 3

  val rdd11: RDD[Int] = sc.makeRDD(0 to 100, 5)rdd11.repartition(3)

12、创建一个 RDD数据为1,3,4,10,4,6,9,20,30,16,请给RDD进行分别进行升序和降序排列

    val rdd12: RDD[Int] = sc.makeRDD(List(1, 3, 4, 10, 4, 6, 9, 20, 30, 16))val resultTwelve: List[Int] = rdd12.sortBy(x => x).collect().toList
//    println(resultTwelve)

13、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，求并集

    val rdd13: RDD[Int] = sc.makeRDD(1 to 6)val rdd1301: RDD[Int] = sc makeRDD (1 to 10)val resultThirteen: List[Int] = rdd13.union(rdd1301).collect().toList
//    println(resultThirteen)

14、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算差集，两个都算

    val rdd14: RDD[Int] = sc.makeRDD(1 to 6)val rdd1401: RDD[Int] = sc makeRDD (4 to 10)val resultFourteen: List[Int] = rdd14.subtract(rdd1401).collect().toList
//    println(resultFourteen)

15、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算交集

   val rdd15: RDD[Int] = sc.makeRDD(1 to 6)val rdd1501: RDD[Int] = sc makeRDD (4 to 10)val resultfifteen: List[Int] = rdd14.intersection(rdd1501).collect().toList
//    println(resultfifteen)

16、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 6和4 to 10，计算 2 个 RDD 的笛卡尔积

    val rdd16: RDD[Int] = sc.makeRDD(1 to 6)val rdd1601: RDD[Int] = sc makeRDD (4 to 10)val resultSixteen: List[(Int, Int)] = rdd16.cartesian(rdd1601).collect().toList
//    println(resultSixteen)

17、创建两个RDD，分别为rdd1和rdd2数据分别为1 to 5和11 to 15，对两个RDD拉链操作

    val rdd17: RDD[Int] = sc.makeRDD(1 to 5)val rdd1701: RDD[Int] = sc.makeRDD(11 to 15)val resultSevenTeen: List[(Int, Int)] = rdd17.zip(rdd1701).collect().toListprintln(resultSevenTeen)

18、创建一个RDD数据为List((“female”,1),(“male”,5),(“female”,5),(“male”,2))，请计算出female和male的总数分别为多少

    val rdd18: RDD[(String, Int)] = sc.makeRDD(List(("female", 1), ("male", 5), ("female", 5), ("male", 2)))val eighteen: List[(String, Int)] = rdd18.groupByKey().map(x => (x._1, x._2.size)).collect().toListprintln(eighteen)

19、创建一个有两个分区的 RDD数据为List((“a”,3),(“a”,2),(“c”,4),(“b”,3),(“c”,6),(“c”,8))，取出每个分区相同key对应值的最大值，然后相加

    val rdd19: RDD[(String, Int)] = sc.makeRDD(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)),2)val nineteen: RDD[(String, Int)] = rdd19.aggregateByKey(0)(math.max(_, _), _ + _)

20、创建一个有两个分区的 pairRDD数据为Array((“a”, 88), (“b”, 95), (“a”, 91), (“b”, 93), (“a”, 95), (“b”, 98))，根据 key 计算每种 key 的value的平均值

    val rdd20: RDD[(String, Int)] = sc.makeRDD(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)val value: RDD[(String, Double)] = rdd20.groupByKey().map(x => (x._1, (x._2.sum / x._2.size).toDouble))
//    println(value.collect().toList)

21、统计出每一个省份广告被点击次数的 TOP3，数据在access.log文件中

*数据结构：时间戳，省份，城市，用户，广告字段使用空格分割。
样本如下：
1516609143867 6 7 64 16
1516609143869 9 4 75 18
1516609143869 1 7 87 12

val rdd21: RDD[String] = sc.textFile("input/access.log")//先统计每个省份广告的被点击次数   (省份 -> 广告） 1,val provinceAdToOne = rdd21.map { x =>val fields: Array[String] = x.split(" ")((fields(1), fields(4)), 1)}//计算每个省中每个广告被点击的总数：((省份,广告),总和)val provinceAdToSum = provinceAdToOne.reduceByKey(_ + _)//将省份作为key，广告加点击数为value：(省份,(广告,总和))val provinceToAdSum = provinceAdToSum.map(x => (x._1._1, (x._1._2, x._2)))//将同一个省份的所有广告进行聚合(省份,List((广告,总和),(广告,总和)...))val provinceGroup = provinceToAdSum.groupByKey()//对同一个省份所有广告的集合进行排序并取前3条，排序规则为广告点击总数,如果需要按照省份ID排序,代码结尾可以添加".sortByKey()"val provinceAdTop3 = provinceGroup.mapValues { x =>x.toList.sortWith((x, y) => x._2 > y._2).take(3)}//将数据拉取到Driver端并打印provinceAdTop3.collect().foreach(println)

22、读取本地文件words.txt,统计出每个单词的个数，保存数据到 hdfs 上

val resultTwentyOne: RDD[(String, Int)] = sc.textFile("word").flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)
//    resultTwentyOne.saveAsTextFile("hdfs://192.168.100.201:8020/wordCount")
//    println(resultTwentyOne.collect().toList)

23、读取 people.json 数据的文件, 每行是一个 json 对象，进行解析输出

  val file3 = sc.textFile("json/people.json")val result: RDD[Option[Any]] = file3.map(JSON.parseFull)
//    println(result.collect().toList)

24、保存一个 SequenceFile 文件，使用spark创建一个RDD数据为Array((“a”, 1),(“b”, 2),(“c”, 3))，保存为SequenceFile格式的文件到hdfs上

    val rdd24 = sc.makeRDD(Array(("a", 1), ("b", 2), ("c", 3)))
//    rdd24.saveAsSequenceFile("hdfs://192.168.100.201:8020/sequenceFile")

25、读取24题的SequenceFile 文件并输出

    val rdd25 = sc.sequenceFile[String,Int]("hdfs://192.168.100.201:8020/sequenceFile/part-00001")val resultTwentyFive: List[(String, Int)] = rdd25.collect().toList
//        println(resultTwentyFive)

26、读写 objectFile 文件，把 RDD 保存为objectFile，RDD数据为Array((“a”, 1),(“b”, 2),(“c”, 3))，并进行读取出来

    val rdd26 = sc.makeRDD(Array(("a", 1), ("b", 2), ("c", 3)))
//    rdd26.saveAsObjectFile("outputObject")val rdd2601 = sc.textFile("outputObject")
//    println(rdd2601.collect().toList)

27、使用内置累加器计算Accumulator.txt文件中空行的数量

    val rdd27: RDD[String] = sc.textFile("Accumulator")var count = sc.longAccumulator("count")rdd27.foreach { x => if (x == "") count.add(1) }println(count.value)

28、使用Spark广播变量

用户表：
id name age gender(0|1)
001,刘向前,18,0
002,冯剑,28,1
003,李志杰,38,0
004,郭鹏,48,2
要求，输出用户信息，gender必须为男或者女，不能为0,1
使用广播变量把Map(“0” -> “女”, “1” -> “男”)设置为广播变量，最终输出格式为
001,刘向前,18,女
003,李志杰,38,女
002,冯剑,28,男
004,郭鹏,48,男

   val rdd28: RDD[String] = sc.textFile("user")val broadcastMap: Broadcast[Map[String, String]] = sc.broadcast(Map(("0" -> "女"), ("1" -> "男")))def broadcastDef(sex:String): String ={if (sex =="0" || sex=="1")broadcastMap.value(sex)else"未知性别"}

29、mysql创建一个数据库bigdata0407，在此数据库中创建一张表

CREATE TABLE user (
id int(11) NOT NULL AUTO_INCREMENT,
username varchar(32) NOT NULL COMMENT ‘用户名称’,
birthday date DEFAULT NULL COMMENT ‘生日’,
sex char(1) DEFAULT NULL COMMENT ‘性别’,
address varchar(256) DEFAULT NULL COMMENT ‘地址’,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
数据如下：
依次是：姓名生日性别省份
安荷 1998/2/7 女江苏省
白秋 2000/3/7 女天津市
雪莲 1998/6/7 女湖北省
宾白 1999/7/3 男河北省
宾实 2000/8/7 男河北省
斌斌 1998/3/7 男江苏省
请使用spark将以上数据写入mysql中，并读取出来。

package com.czxy.homeworkOneimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.{JdbcRDD, RDD}object demo29 {def main(args: Array[String]): Unit = {//1.设置Spark的运行环境val config = new SparkConf().setMaster("local[*]")setAppName("sparkConf")//2.创建spark的上下文对象， 与clusterManager进行通信。分配管理任务val sc: SparkContext = new SparkContext(config)//准备一些数据，插入mysqlval data: RDD[(String, String, String, String)] = sc.makeRDD(List(("安荷", "1998/2/7", "女", "河北省"), ("白秋", "2000/3/7", "女", "天津市"), ("雪莲", "1998/6/7", "女", "湖北省"), ("宾白", "1998/6/7", "女", "河北省"), ("宾实", "2000/8/7", "男", "河北省"), ("斌斌", "1998/3/7", "男", "江苏省")))data.foreachPartition(insertToMysql)//    checkData(sc)}def checkData(sc:SparkContext)={//设置与mysql的连接//第一个参数：指定需要连接的数据库//第二个参数:用户名//第三个参数：密码def getConn(): Connection = {DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")}//通过连接获取数据//第一个参数：获取与spark上下文的连接//第二个参数：建立到给定数据库URL的连接//第三个参数：需要执行的sql语句//第四个参数：第一个占位符的最小值//第五个参数：第二个占位符的最大值//第六个参数：分区数量//第七个参数：获取到从ResultSet到所需结果类型的单行的函数。var sql = "select * from user where id >= ? and id <= ?;"val infoRDD: JdbcRDD[(Int, String, String,String,String)] =  new JdbcRDD(sc,getConn,sql,4,6,2,rs =>{val id: Int = rs.getInt("id")val name: String = rs.getString("name")val birthday: String = rs.getString("birthday")val sex: String = rs.getString("sex")val address: String = rs.getString("address")(id,name,birthday,sex,address)})println(infoRDD.collect().toList)}/*** 插入数据到mysql中* @param partitionData*/def insertToMysql(partitionData:Iterator[(String,String,String,String)]) ={//将数据存入到Mysql//获取连接val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")//遍历每个分区内的数据，把数据插入到mysql中partitionData.foreach(data => {val sql ="INSERT INTO `user` VALUES (NULL, ?, ?, ?, ?);"//执行sqlval ps: PreparedStatement = conn.prepareStatement(sql)ps.setString(1,data._1)ps.setString(2,data._2)ps.setString(3,data._3)ps.setString(4,data._4)//提交并执行ps.execute()})//关闭连接conn.close()}}

30、在hbase中创建一个表student，有一个 message列族

create ‘student’, ‘message’
scan ‘student’, {COLUMNS => ‘message’}
给出以下数据，请使用spark将数据写入到hbase中的student表中,并进行查询出来
数据如下：
依次是：姓名班级性别省份，对应表中的字段依次是：name,class,sex,province
飞松 3 女山东省
刚洁 1 男深圳市
格格 4 女四川省
谷菱 5 女河北省
国立 2 男四川省
海涛 3 男广东省
含芙 3 女四川省
华敏 4 女上海市
乐和 2 男上海市
乐家 3 男黑龙江
乐康 4 男湖北省
乐人 5 男四川省
乐水 3 男北京市
乐天 4 男河北省
乐童 5 男江苏省
乐贤 1 男陕西省
乐音 2 男广东省
李仁 3 男湖北省
立涛 3 女陕西省
凌青 4 女湖北省
陆涛 4 男山东省
媚媚 5 女河南省
梦亿 4 男江苏省
铭忠 5 男四川省
慕梅 3 女北京市
鹏吉 1 男上海市
娉婷 4 女河南省
淇峰 2 男广东省
庆元 3 男上海市
庆滋 4 男北京市
丘东 5 男江苏省
荣郑 1 男黑龙江
蕊蕊 5 女四川省
尚凯 2 男北京市
诗涵 1 女河南省
淑凤 2 女天津市
淑娇 3 女上海市
淑燕 4 女河北省
淑怡 4 女广东省
思璇 2 女湖北省
苏华 3 女山东省
苏梅 4 女四川省
听荷 5 女深圳市
文怡 1 女天津市
文怡 2 女河北省
香凝 3 女山东省
翔云 4 女河南省
小芸 5 女深圳市

package com.czxy.homeworkOneimport org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client.{HBaseAdmin, Put, Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDDobject demo30 {def main(args: Array[String]): Unit = {//1.设置Spark的运行环境val config = new SparkConf().setAppName("DataSourceTest").setMaster("local[*]")//2.创建spark的上下文对象， 与clusterManager进行通信。分配管理任务val sc = new SparkContext(config)//设置日志级别sc.setLogLevel("WARN")////创建  HBase所需的环境配置val conf = HBaseConfiguration.create()//设置连接conf.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181")//设置需要操作的表val studentTable = TableName.valueOf("student")//获取表的描述val tableDescr = new HTableDescriptor(studentTable)//添加列族tableDescr.addFamily(new HColumnDescriptor("message".getBytes))//创建Hbase的管理对象val admin = new HBaseAdmin(conf)//如果表已存在就先禁用表，然后删除表if (admin.tableExists(studentTable)) {admin.disableTable(studentTable)admin.deleteTable(studentTable)}//创建表admin.createTable(tableDescr)//给列族添加列def convert(triple: (String, String, String,String)) = {val put = new Put(Bytes.toBytes(triple._1))put.addImmutable(Bytes.toBytes("message"), Bytes.toBytes("classes"), Bytes.toBytes(triple._2))put.addImmutable(Bytes.toBytes("message"), Bytes.toBytes("sex"), Bytes.toBytes(triple._3))put.addImmutable(Bytes.toBytes("message"), Bytes.toBytes("address"), Bytes.toBytes(triple._4))(new ImmutableBytesWritable, put)}//准备一批数据添加到表中//    val dataRDD: RDD[(String, String, String)] = sc.parallelize(List(("1","apple","11"), ("2","banana","12"), ("3","pear","13")))val rdd30: RDD[Array[String]] = sc.textFile("info").map(_.split("\t"))val list: List[List[(String, String, String, String)]] = rdd30.map(x => List((x(0), x(1), x(2), x(3)))).collect().toListval flatten: List[(String, String, String, String)] = list.flattenval dataRDD: RDD[(String, String, String, String)] = sc.parallelize(flatten)val targetRDD: RDD[(ImmutableBytesWritable, Put)] = dataRDD.map(convert)//构造一个map/reduce得作业配置val jobConf = new JobConf(conf)//设置mapreduce 的输出位置jobConf.setOutputFormat(classOf[TableOutputFormat])jobConf.set(TableOutputFormat.OUTPUT_TABLE, "student")//使用Hadoop JobConf对象将RDD输出到任何Hadoop支持的存储系统targetRDD.saveAsHadoopDataset(jobConf)println("写入数据成功")//读取数据conf.set(TableInputFormat.INPUT_TABLE, "student")//使用InputFormat获取给定Hadoop文件的RDD//和传递到输入格式的额外配置选项。val hbaseRDD: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],classOf[org.apache.hadoop.hbase.client.Result])//返回rdd的数量val count: Long = hbaseRDD.count()println("hBaseRDD RDD Count:"+ count)//遍历查询表中的数据hbaseRDD.foreach {case (_, result) =>val key = Bytes.toString(result.getRow)val classes = Bytes.toString(result.getValue("message".getBytes, "classes".getBytes))val sex = Bytes.toString(result.getValue("message".getBytes, "sex".getBytes))val address = Bytes.toString(result.getValue("message".getBytes, "address".getBytes))println("Row key:" + key + ", Name:" + key + ", class:" + classes+", address:"+address)}sc.stop()}}

好了，以上内容到这里就结束了。各位的点赞、关注就是小编坚持下去的动力。我们下期见哦~~~

我是小哪吒。一个互联网行业的业余选手 …

贪图省力的船夫，目标永远下游