PySpark的不同文件格式读取，如：sc.parallelize、hdfs读取资料、csv、txt 、json
## 用pysaprk建立第一个RDD
from __future__ import print_function,division
from pyspark import SparkConf,SparkContext
from pysaprk.sql import SparkSession
## 启动spark
spark= SparkSession.builder.master(‘local’).appName(‘test’).getOrCreate()
sc = spark.sparkContext
## 建立第一个RDD --- sparkContext
wordsList = ['cat','elephant','rat','cat']
wordsRDD = sc.parallelize(wordsList,2) #parallelize
print(type(wordsRDD)) #查看类型
wordsRDD.count() #查看行数
wordsRDD.take(5) #前5个项目
wordsRDD.collect() #适用于小数据量
## 从hdfs中读取资料
textFromHDFS = spark.read.text('hdfs://tmp/NASA_access') #读取文件
print(type(textFromHDFS)) #查看类型
textFromHDFS.head() #查看前几行 -- 推荐使用
textFromHDFS.show() #查看资料
## 如何从csv -- 读取 ---转换dataframe
!hadoop fs -tail /tmp/rating.csv #查看前几行数据
##读取数据的参数设置
path = 'hdfs://tmp/rating.csv'
schema = None
sep = ','
header = True
csvDF = spark.read.csv(path = path, schema = schema ,sep = sep, header = header) #读取数据集
print(type(csvDF)) #查看类型
print(csvDF)
csvDF.head() #查看前几行 -- 推荐使用
csvDF.show() #查看资料
## 读取txt
textDF = spark.read.text(paths = path)
textDF.head() #查看前几行 -- 推荐使用
textDF.show() #查看资料
## 读取json文件
jsonDF = spark.read.json('hdfs://tmp/json_example.json')
jsonDF.head() #查看前几行 -- 推荐使用
jsonDF.show() #查看资料
PySpark的不同文件格式读取，如：sc.parallelize、hdfs读取资料、csv、txt 、json## 用pysaprk建立第一个RDDfrom __future__ import print_function,divisionfrom pyspark import SparkConf,SparkContextfrom pysaprk.sql import SparkSes...
使用python
读取
和保存为excel、
csv
、
txt
文件以及对DataFrame文件的基本操作
读取
excel文件并将其内容转化为矩阵形式。
对DataFrame文件的基本操作包括DataFrame的创建（dict进行创建，也可以
读取
csv
或者
txt
文件）、DataFrame轴的概念和DataFrame一些性质（索引、切片；修改数据；算数运算；函数应用和映射；排序和排名；汇总和计算描述统计；处理缺失数据
Dataframe中的Series是什么？......
1.常见的
读取
数据源
文件格式
和文件系统。对于存储在本地文件系统或分布式文件系统（比如
HDFS
）中的数据，Spark可以访问很多种
不同
的
文件格式
。包括文本文件、JSON、SequenceFile、以及protocol buffer。
Spark SQL中的结构化数据源。
数据库和键值存储。Spark自带的库以及一些第三方库，可以用来连接HBase、JDBC源。
文章目录
pyspark
读取
数据参数介绍formatschemaloadtableoption
读取
文件json
csv
parquet和orc
读取
数据表hivejdbc
pyspark
读取
数据
format
DataFrameReader.format(source)
指定
不同
格式来
读取
文件，例如以下为指定json格式
读取
数据：
df = spark.read.format('json').load('python/test_support/sql/people.json')
针对常用几个
文件格式
网上找了很多
资料
，例如 textFile 和wholeTextFiles方法。但这两个都是都是
读取
txt
文件，然后生成RDD的格式。
如果再从 RDD 转为
Pyspark
的 DataFrame格式，那么就多此一举了。
终于找到了直接
读取
txt
文件并生成 DataFrame的方法，那就是 spark 的 read.
txt
方法。
实现代码如下：
from
pyspark
.sql i...
from
pyspark
.sql.types import *
schema = StructType([
StructField("id", LongType(), True),
StructField("name", StringType(), True),
collect是
pyspark
中可以将rdd转换为list的转换操作，虽然很好用，但也需要注意，不是可以随心所欲地用的。
collect的
读取
相当于从所有分布式机器上把数据拉下来放在本地展示：
这个操作一方面把分布式变成了单机操作，失去了分布式的意义；
另一方面就是存放本地会消耗相当一部分的内存；当rdd很大时，内存溢出会直接导致程序卡死。
所以如果只想看看数据格式，用take取样就可以了...
hadoop@rachel-virtual-machine:/usr/local/spark$ ./bin/
pyspark
./bin/
pyspark
: 行 45: python: 未找到命令
Python 3.6.8 (default, Jan 14 201...
<h3>回答1：</h3><br/>使用
pyspark
读取
hdfs
的
csv
文件可以按照以下步骤进行：
1. 首先，需要创建一个SparkSession对象，可以使用以下代码：
from
pyspark
.sql import SparkSession
spark = SparkSession.builder.appName("Read
CSV
from
HDFS
").getOrCreate()
2. 接下来，使用SparkSession对象的read方法
读取
csv
文件，可以使用以下代码：
df = spark.read.
csv
("
hdfs
://<namenode>:<port>/<path_to_file>.
csv
", header=True, inferSchema=True)
其中，<namenode>是
HDFS
的NameNode节点，<port>是
HDFS
的端口号，<path_to_file>是
csv
文件在
HDFS
中的路径。
3. 最后，可以使用DataFrame对象的方法对数据进行处理和分析，例如：
df.show()
这将显示DataFrame中的前20行数据。
完整的代码示例：
from
pyspark
.sql import SparkSession
spark = SparkSession.builder.appName("Read
CSV
from
HDFS
").getOrCreate()
df = spark.read.
csv
("
hdfs
://<namenode>:<port>/<path_to_file>.
csv
", header=True, inferSchema=True)
df.show()
注意：在使用
pyspark
读取
hdfs
的
csv
文件时，需要确保Hadoop和Spark的环境变量已经设置好。
<h3>回答2：</h3><br/>
PySpark
是Apache Spark的一个Python API，可以用于分布式数据处理和大规模数据分析。
HDFS
是Hadoop分布式文件系统，一般用于存储大规模数据。
PySpark
可以
读取
HDFS
上的
CSV
文件进行数据处理和分析。
首先需要在
PySpark
中导入所需的库：
```python
from
pyspark
.sql import SparkSession
然后创建一个SparkSession实例：
```python
spark = SparkSession.builder \
.appName("Read
CSV
from
HDFS
") \
.config("spark.hadoop.fs.defaultFS", "
hdfs
://namenode:9000") \
.getOrCreate()
其中，appName用于设置应用程序的名称，config用于设置
HDFS
的地址和端口号，getOrCreate方法用于获取现有的SparkSession实例或者创建一个新的实例。
接下来可以使用SparkSession的read方法来
读取
CSV
文件：
```python
df = spark.read \
.option("header", "true") \
.option("inferSchema", "true") \
.
csv
("/path/to/
csv
/file.
csv
")
其中，option方法用于设置
读取
CSV
文件的参数，如header表示文件包含列名，inferSchema表示自动推断列的类型，
csv
方法用于指定
CSV
文件的路径。
最后可以使用DataFrame API对数据进行处理和分析：
```python
df.show()
df.groupBy("column_name").count().show()
df.select("column_name").distinct().count()
其中，show方法用于显示前几条数据，groupBy方法用于分组统计数据，distinct方法用于去重统计数据。
总而言之，
PySpark
可以方便地
读取
HDFS
上的
CSV
文件进行数据处理和分析，同时也提供了丰富的API和方法帮助用户处理大规模数据。
<h3>回答3：</h3><br/>
Pyspark
是一种用于处理大规模数据的 Python 库，它也是 Apache Spark 生态系统的一部分。在
Pyspark
中，我们可以使用 SparkSession API 来
读取
和写入数据。而在 Hadoop 分布式文件系统 (
HDFS
) 上存储大量的数据文件，
CSV
格式是其中最常见的一种。
Pyspark
通过 Hadoop File System 将
HDFS
上的
CSV
文件
读取
到 Spark 数据结构中，
读取
后的数据可以转换为 DataFrame 或者 RDD。下面是
Pyspark
读取
HDFS
的
CSV
文件的一些详细步骤：
1. 首先需要创建一个 SparkSession 对象，可以使用下面的示例代码：
from
pyspark
.sql import SparkSession
spark = SparkSession.builder.appName("Read
CSV
").getOrCreate()
2. 导入
CSV
文件：
data = spark.read.format("
csv
").option("header", "true").load("
hdfs
://path/to/
csv
/file.
csv
")
在这个例子中，我们使用 `spark.read` API 创建 DataFrame，
读取
CSV
数据文件。`format` 函数指定
读取
的
文件格式
，这里我们指定为
CSV
。`option` 函数用来传递
读取
CSV
文件时需要的一些选项，如 `header` 表示数据中是否包括表头。`load` 函数用于指定要
读取
的文件路径。
3. 数据处理：
data.show()
data.printSchema()
这个例子中我们展示 DataFrame 内容，以及数据结构。
4. 关闭 SparkSession：
spark.stop()
Pyspark
的
读取
操作具有极高的可扩展性和并行性，能够对 TB 级别的数据进行
读取
处理。因此，使用
Pyspark
读取
HDFS
的
CSV
文件非常适合于大规模数据的处理和分析场景。
