TensorFlow入门8 -- 导入数据之Dataset机制

Posted 2023-04-27

参考技术A 参考：《 深度学习图像识别技术 --基于TensorFlow Object Detection API 和 OpenVINO 》

TensorFlow用 tf.data  API 实现数据导入。输入数据流（ input pipelines）可以是图像，也可以是文字（text）。

tf.data  API 定义了 两个抽象类型 ：

一个是 tf.data.Dataset 类 ，表示元素序列。每一个元素包含一个或多个Tensor对象。例如，在一个图像数据流（image pipeline）中, 一个元素可能是一个训练数据实例（training example）,该实例有一对 tensors 分别是图像数据( image data)和标签数据(label) 。

        有两种显示创建dataset的方式：

        通过从一个或多个tf.Tensor 对象上创建一个 源 ( source ) (例如，用 Dataset.from_tensor_slices()) 来构造一个 dataset 

         通过对一个或多个tf.data.Dataset对象执行 转换 (  transformation ) (例如，用 Dataset.batch()) 来构造一个 dataset 

另一个是 tf.data.Iterator 类， 它提供从数据集中提取元素的方法。 Iterator.get_next() 返回Dataset的下一个元素，它通常充当输入数据流（input pipeline）和模型（model）之间的接口。最简单的迭代器（iterator）是一个“单次迭代器（one-shot iterator）”，它与特定的Dataset相关联并遍历一次。 对于更复杂的用途，  Iterator.initializer 操作允许您使用不同的数据集重新初始化和参数化迭代器，以便您可以在同一个程序中多次迭代训练和验证数据。

要启动输入管道（input pipeline），必须定义一个 源（Source）。 例如，要从内存中的张量构建Dataset ，可以使用 tf.data.Dataset.from_tensors() 或 tf.data.Dataset.from_tensor_slices() 。又例如，如果输入数据以建议的 TFRecord格式 存储在磁盘上，则可以构建一个 tf.data.TFRecordDataset  。

一旦有了Dataset对象，就可以通过tf.data.Dataset对象上的一系列方法调用将其转换为新的满足训练模型需要的Dataset 。 例如，用 Dataset.map() （将函数应用于每个元素）实现每一个元素的转换；用Dataset.batch()实现多元素转换 。 有关转换的完整列表，请参阅 tf.data.Dataset 的文档。范例程序如下：

使用数据集（Dataset）中最常用的方法是创建一个 迭代器（iterator） 对象，该对象提供对数据集元素的访问。（例如，通过调用 Dataset.make_one_shot_iterator() 创建一个单次迭代器）。 tf.data.Iterator提供了两个操作：  Iterator.initializer  ，实现初始化或重新初始化迭代器的状态; 和 Iterator.get_next()  ，它返回下一个元素的tf.Tensor对象。 根据不同的使用情况，选择不同类型的迭代器。

大数据入门之Spark快速入门及导入数据，求平均值

运行环境

本文的具体运行环境如下：

• CentOS 7.6
• Spark 2.4
• Hadoop 2.6.0
• Java JDK 1.8
• Scala 2.10.5

一、下载安装

首先在官网

https://spark.apache.org/downloads.html

下载对应版本的Spark

丢到你的服务器上 自己的路径 比如 /user/hadoop/My_Spark

解压 

　　tar -xvf XXX.tar.gz(你的压缩包名称)

然后 记录你的 路径  /user/hadoop/My_Spark/spark-2.4.0-bin-hadoop2.7

配置spark用户权限

　　sudo chown -R hadoop:hadoop ./spark # 此处的 hadoop 为你的用户名 ./spark为你的路径名

安装后，需要在 ./conf/spark-env.sh 中修改 Spark 的 Classpath，执行如下命令拷贝一个配置文件：

1. cd /user/hadoop/My_Spark/spark-2.4.0-bin-hadoop2.7
2. cp ./conf/spark-env.sh.template ./conf/spark-env.sh

编辑 ./conf/spark-env.sh（vim ./conf/spark-env.sh） ，在最后面加上如下一行：

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)

保存后，Spark 就可以启动运行了。

当然 还少不了设置环境变量

　　vi ~/.bash_profile

在最后加入

　　# spark
　　export SPARK_HOME= （你的Spark路径）
　　export PATH=$PATH:$SPARK_HOME/bin

运行 Spark 示例

注意，必须安装 Hadoop 才能使用 Spark，但如果使用 Spark 过程中没用到 HstudentS，不启动 Hadoop 也是可以的。此外，接下来教程中出现的命令、目录，若无说明，则一般以 Spark 的安装目录（/usr/local/spark）为当前路径，请注意区分。

在 ./examples/src/main 目录下有一些 Spark 的示例程序，有 Scala、Java、Python、R 等语言的版本。我们可以先运行一个示例程序 SparkPi（即计算 π 的近似值），执行如下命令：

1. cd /user/hadoop/My_Spark/spark-2.4.0-bin-hadoop2.7 #你的路径
   
2. ./bin/run-example SparkPi

通过 Spark Shell 进行交互分析

Spark shell 提供了简单的方式来学习 API，也提供了交互的方式来分析数据。Spark Shell 支持 Scala 和 Python，本教程选择使用 Scala 来进行介绍。

Scala

Scala 是一门现代的多范式编程语言，志在以简练、优雅及类型安全的方式来表达常用编程模式。它平滑地集成了面向对象和函数语言的特性。Scala 运行于 Java 平台（JVM，Java 虚拟机），并兼容现有的 Java 程序。

Scala 是 Spark 的主要编程语言，如果仅仅是写 Spark 应用，并非一定要用 Scala，用 Java、Python 都是可以的。使用 Scala 的优势是开发效率更高，代码更精简，并且可以通过 Spark Shell 进行交互式实时查询，方便排查问题。

执行如下命令启动 Spark Shell：

1. ./bin/spark-shell

Shell 命令

启动成功后如图所示，会有 “scala >” 的命令提示符。

成功启动Spark Shell

基础操作

Spark 的主要抽象是分布式的元素集合（distributed collection of items），称为RDD（Resilient Distributed Dataset，弹性分布式数据集），它可被分发到集群各个节点上，进行并行操作。RDDs 可以通过 Hadoop InputFormats 创建（如 HstudentS），或者从其他 RDDs 转化而来。

 我们从本地路径读取一个预先准备好的student.txt文件

student.txt 文件内容如下 有四个字段 id name age score

在交互式窗口中输入　　

　　

import org.apache.spark.sql.types._
import org.apache.spark.sql.SparkSession
val spark=SparkSession.builder().getOrCreate()
val student = spark.read.option("header", true).option("sep",",").schema("id int,name string,age int,score int").csv("file:///u01/isi/app/hadoop/run/student.txt")

这里student是 spark的一个DataFrameReader 也就是RDD DataFrame 通过读取本地文件获得的。

整个相当于一个临时的表

我们这里以CSV格式读入 分隔符为 , 然后首行开启  .schema表示以指定的模式读入

代码中通过 “file://” 前缀指定读取本地文件。Spark shell 默认是读取 HstudentS 中的文件，需要先上传文件到 HstudentS 中，否则会有“org.apache.hadoop.mapred.InvalidInputException: Input path does not exist:”的错误。

 

 读入操作详解

API地址

DataFrameReader支持若干种操作方式 可使用.option在后面具体加入参数。

截取部分参数如下

 

sep (default ,)	设置单个字符作为每个字段和值的分隔符。
encoding (default UTF-8)	根据给定的编码类型解码CSV文件。
quote (default ")	设置用于转义引号值的单个字符，分隔符可以是该值的一部分。如果要关闭引号，需要设置的不是null，而是一个空字符串。此行为与com. database .spark.csv不同。
escape (default \\)	设置一个字符，用于转义已引用值中的引号
charToEscapeQuoteEscaping (default escape or \\0)	设置用于转义引号字符的单个字符。当转义和引号字符不同时，默认值为转义字符，否则为\\0。
comment (default empty string)	:设置一个字符，用于跳过以该字符开头的行。默认情况下，它是禁用的。
header (default false)	使用第一行作为列的名称。

DataFrame基本动作运算

show展示数据

可以用show() 方法来展示数据，show有以下几种不同的使用方式：
show()：显示所有数据
show(n) ：显示前n条数据
show(true): 最多显示20个字符，默认为true
show(false): 去除最多显示20个字符的限制
show(n, true）：显示前n条并最多显示20个自负

student.show()
student.show(3)
student.show(true)
student.show(false)
student.show(3,true)

 

输入

student.show(student.count().toInt)

 

按成绩倒序排序输出

输入

 student.sort(student("score").desc).show(student.count().toInt)

describe(cols: String*)：获取指定字段的统计信息

这个方法可以动态的传入一个或多个String类型的字段名，结果仍然为DataFrame对象，用于统计数值类型字段的统计值，比如count, mean, stddev, min, max等。

 

 

求平均分数并输出

student.agg(mean("score")).show
//或者下面这种 两种方式都可以
student.describe("score").where("summary =\'mean\'").show()

 

 

 

单个DataFrame操作

 使用where筛选条件

where(conditionExpr: String)：SQL语言中where关键字后的条件 ，传入筛选条件表达式，可以用and和or。得到DataFrame类型的返回结果， 比如我们想得到用户1或者使用助手1的操作记录：

student.where("user=1 or type =\'助手1\'").show()

或者如上图 

student.describe("score").where("summary =\'mean\'").show()

select：获取指定字段值

根据传入的String类型字段名，获取指定字段的值，以DataFrame类型返回,比如我们想要查找user和type两列：

student.select("user","type").show()

 

好了，本次教程到此结束。剩下的自己看文档吧。

 

退出 

:quit