8.spark core之读写数据

发布时间:2020-05-30 23:56:46 作者:菲立思教育
来源:网络 阅读:11115

  spark支持多种数据源,从总体来分分为两大部分:文件系统和数据库。

文件系统

  文件系统主要有本地文件系统、Amazon S3、HDFS等。

  文件系统中存储的文件有多种存储格式。spark支持的一些常见格式有:

格式名称 结构化 说明
文件文件 普通文件文件,每行一条记录
JSON 半结构化 常见的基于文本的半结构化数据
CSV 常见的基于文本的格式,在电子表格应用中使用
SequenceFiles 一种用于键值对数据的常见Hadoop文件格式

文本文件

  输出文本文件时,可使用saveAsTextFile()方法接收一个目录,将RDD中的内容输出到目录中的多个文件中。

```
result.saveAsTextFile(outputFile)
```

JSON

CSV与TSV

  CSV与TSV文件每行都有固定的字段,字段之间使用分隔符(CSV使用逗号;tsv使用制表符)分隔。

SequenceFile

  SequenceFile是键值对形式的常用Hadoop数据格式。由于Hadoop使用一套自定义的序列化框架,因此SequenceFile的键值对类型需实现Hadoop的Writable接口。

数据库

  数据库主要分为关系型数据库MySQL、PostgreSQL等)和非关系型数据库(HBase、ElasticSearch等)。

JDBC数据库连接

  spark使用JDBC访问关系型数据库(MySQL、PostgreSQL等),只需要构建一个org.apache.spark.rdd.JdbcRDD即可。

def createConnection() = {
    Class.forName("com.mysql.jdbc.Driver").newInstance()
    DriverManager.getConnection("jdbc:mysql://localhost/test", "root", "root")
}

def extractValues(r: ResultSet) = {
    (r.getInt(1), r.getString(2))
}

val data = new JdbcRDD(sc, createConnection, 
                "SELECT * FROM panda WHERE id >= ? AND id <= ?"),
                lowerBound = 1, upperBound = 3, 
                numPartitions = 2, mapRow = extractValues)
println(data.collect().toList)

HBase

  spark通过Hadoop输入格式(org.apache.hadoop.hbase.mapreduce.TableInputFormat)访问HBase。这种格式返回键值对数据,键类型为org.apache.hadoop.hbase.io.ImmutableBytesWritable,值类型为org.apache.hadoop.hbase.client.Result。

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat

val conf = HBaseConfiguration.create()
conf.set(TableInputFormat.INPUT_TABLE, "tablename")

val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], ClassOf[Result])

ElasticSearch

  spark使用ElasticSearch-Hadoop连接器从ElasticSearch中读写数据。ElasticSearch连接器依赖于SparkContext设置的配置项。ElasticSearch连接器也没有用到Spark封装的类型,而使用saveAsHadoopDataSet。

def mapWritableToInput(in: MapWritable): Map[String, String] = {
    in.map{case (k, v) => (k.toString, v.toString)}.toMap
}

val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set(ConfigurationOptions.ES_RESOURCE_READ, args[1])
jobConf.set(ConfigurationOptions.ES_NODES, args[2])

val currentTweets = sc.hadoopRDD(jobConf, classOf[EsInputFormat[Object, MapWritable]], classOf[Object], ClassOf[MapWritable])

val tweets = currentTweets.map{ case (key, value) => mapWritableToInput(value) }
val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set("mapred.output.format.class", "org.elasticsearch.hadoop.mr.EsOutFormat")
jobConf.setOutputCommitter(classOf[FileOutputCommitter])
jobConf.set(ConfigurationOptions.ES_RESOURCE_WRITE, "twitter/tweets")
jobConf.set(ConfigurationOptions.ES_NODES, "localhost")
FileOutputFormat.setOutputPath(jobConf, new Path("-"))
output.saveAsHadoopDataset(jobConf)

忠于技术,热爱分享。欢迎关注公众号:java大数据编程,了解更多技术内容。

8.spark core之读写数据

推荐阅读:
  1. 【全集】IDEA入门到实战
  2. Spark SQL中怎么创建DataFrames

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

spark csv json

上一篇:MongoDB入门3——更新文档和修改器1

下一篇:数据结构(八)——栈

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》