球墨铸铁复合树脂水箅 小编整理了以下信息供大家参考。
点击此处下载标准CJ/T 328-2010 球墨铸铁复合树脂水箅全文pdf文件。
本标准规定了球墨铸铁复合树脂水算的术语和定义、分类和结构形式、材料、要求、试验方法、检验规则、标志、包装、运输和贮存。本标准适用于安装在各种道路、广场、绿地、厂房车间室内地面等处的球墨铸铁复合树脂水箅。
点击上方关注，All in AI中国
在这篇文章中，我将讨论Apache Spark以及如何在其中创建简单但强大的ETL管道。您将了解Spark如何提供API以将不同的数据格式转换为数据帧和SQL以进行分析，以及如何将一个数据源转换为另一个数据源。
什么是Apache Spark？
根据维基百科：
Apache Spark是一个开源的分布式通用集群计算框架。 Spark提供了一个接口，用于使用隐式数据并行和容错来编程整个集群。
官方网站：
Apache Spark是用于大规模数据处理的统一分析引擎。
简而言之，Apache Spark是一个用于处理、查询和分析大数据的框架。由于计算是在内存中完成的，因此它比MapReduce等竞争对手要快好几倍。以每天产生数TB的数据的速率，需要一种能够以高速提供实时分析的解决方案。一些Spark功能是：
- 它比传统的大规模数据处理框架快100倍。
- 易于使用，因为您可以在Python、R和Scala中编写Spark应用程序。
- 它为SQL、Steaming和Graph计算提供了库。
Apache Spark组件
Spark 核心
它包含Spark的基本功能，如任务调度、内存管理，与存储的交互等。
Spark SQL
它是一组用于与结构化数据交互的库。它使用类似SQL的界面与各种格式的数据进行交互，如CSV、JSON、Parquet等。
Spark Streaming
Spark Streaming是一个Spark组件，支持处理实时数据流。实时流，如股票数据、天气数据、日志和其他各种。
MLIB
MLib是Spark提供的一套机器学习算法，用于监督和无监督学习
GraphX
它是Apache Spark用于图形和图形并行计算的API。它扩展了Spark RDD API，允许我们创建一个带有附加到每个顶点和边缘的任意属性的有向图。它为ETL、探索性分析和迭代图计算提供了统一的工具。
Spark集群管理器
Spark支持以下资源/集群管理器：
- Spark Standalone - Spark附带的简单集群管理器
- Apache Mesos - 一个也可以运行Hadoop应用程序的通用集群管理器。
- Apache Hadoop YARN - Hadoop 2中的资源管理器
- Kubernetes - 一个开源系统，用于自动化容器化应用程序的部署、扩展和管理。
设置和安装
从这里下载Apache Spark的二进制文件。您必须在系统上安装Scala，并且还应设置其路径。
对于本教程，我们使用的是在2019年5月发布的2.4.3版。将文件夹移到/ usr / local中
mv spark-2.4.3-bin-hadoop2.7 / usr / local / spark然后导出Scala和Spark的路径。
#Scala Pathexport PATH="/usr/local/scala/bin:$PATH"#Apache Spark pathexport PATH="/usr/local/spark/bin:$PATH"通过在终端上运行spark-shell命令来调用Spark Shell。如果一切顺利，你会看到如下所示：
它加载基于Scala的shell。由于我们将使用Python语言，因此我们必须安装PySpark。
pip install pyspark安装完成后，您可以通过在终端中运行命令pyspark来调用它：
您找到了一个典型的Python shell，但它加载了Spark库。
用Python开发
让我们开始编写我们的第一个程序。
from pyspark.sql import SparkSessionfrom pyspark.sql import SQLContextif __name__ == \'__main__\': scSpark = SparkSession \\ .builder \\ .appName("reading csv") \\ .getOrCreate()我们导入了两个库：SparkSession和SQLContext。
SparkSession是编写Spark应用程序的入口点。它允许您与Spark提供的DataSet和DataFrame API进行交互。我们通过调用appName来设置应用程序名称。 getOrCreate（）方法返回应用程序的新SparkSession或返回现有的SparkSession。
我们的下一个目标是读取CSV文件。我创建了一个示例CSV文件，名为data.csv，如下所示：
name,age,countryadnan,40,Pakistanmaaz,9,Pakistanmusab,4,Pakistanayesha,32,Pakistan和代码：
if __name__ == \'__main__\': scSpark = SparkSession \\ .builder \\ .appName("reading csv") \\ .getOrCreate()data_file = \'/Development/PetProjects/LearningSpark/data.csv\' sdfData = scSpark.read.csv(data_file, header=True, sep=",").cache() print(\'Total Records = {}\'.format(sdfData.count())) sdfData.show()我设置了文件路径，然后调用.read.csv来读取CSV文件。参数是不言自明的。 .cache（）缓存返回resultset，从而提高性能。当我运行该程序时，它返回如下所示的内容：
看起来有趣，不是吗？现在，如果我想读取数据帧中的多个文件，该怎么办？让我们创建另一个文件，我将其称为data1.csv，它如下所示：
1
2
3
4
五
姓名，年龄，国家
诺琳，23，英格兰
阿米尔，9，巴基斯坦
诺曼，4，巴基斯坦
拉希德，12，巴基斯坦
我只需要这样做：
data_file =\'/Development/PetProjects/LearningSpark/data*.csv\'它将读取所有以CSV类型的数据开头的文件。它将如何读取与模式和转储结果匹配的所有CSV文件：
如您所见，它将CSV中的所有数据转储到单个数据帧中。
但有一点，只有当所有CSV都遵循某种模式时，这种转储才有效。如果您有一个具有不同列名的CSV，那么它将返回以下消息。
19/06/04 18:59:05 WARN CSVDataSource: Number of column in CSV header is not equal to number of fields in the schema: Header length: 3, schema size: 17CSV file: file:///Development/PetProjects/LearningSpark/data.csv如您所见，Spark抱怨不能处理不同的CSV文件。
您可以使用DataFrame执行许多操作，但Spark为您提供了更简单、更熟悉的接口来使用SQLContext操作数据。它是SparkSQL的网关，它允许您使用类似SQL的查询来获得所需的结果。
在我们进一步发展之前，让我们先玩一些真实的数据。为此，我们使用的是从Kaggle得到的超市销售数据。在我们尝试SQL查询之前，让我们尝试按性别对记录进行分组。我们正在处理ETL的提取部分。
data_file = \'/Development/PetProjects/LearningSpark/supermarket_sales.csv\'sdfData = scSpark.read.csv(data_file, header=True, sep=",").cache()gender = sdfData.groupBy(\'Gender\').count()print(gender.show())当你运行时，它会返回如下内容：
groupBy（）按给定列对数据进行分组。在我们的例子中，它是性别列。
SparkSQL允许您使用类似SQL的查询来访问数据。
sdfData.registerTempTable("sales")output = scSpark.sql(\'SELECT * from sales\')output.show()首先，我们从dataframe中创建一个临时表。为此，使用了registerTampTable。在我们的例子中，表名是sales。完成后，您可以在其上使用典型的SQL查询。在我们的例子中，它是Select * from sales。
或者类似下面的内容：
output = scSpark.sql(\'SELECT * from sales WHERE `Unit Price` < 15 AND Quantity < 10\')output.show()甚至是聚合值。
output = scSpark.sql(\'SELECT COUNT(*) as total, City from sales GROUP BY City\')output.show()非常灵活，对吗？
我们刚刚完成了ETL的变换部分。
最后是ETL的加载部分。如果要保存转换后的数据怎么办？您会有很多的可用的选项，RDBMS、XML或JSON。
output.write.format(\'json\').save(\'filtered.json\')运行时，Sparks会创建以下文件夹/文件结构。
它创建了一个具有文件名称的文件夹，在我们的例子中是filtered.json。然后，名为SUCCESStells的文件是否成功运行。如果失败，则生成名为FAILURE的文件。然后，您在此处找到多个文件。多个文件的原因是每个工作都涉及在文件中写入的操作。如果要创建单个文件（不建议使用），则可以使用合并来收集所有分区中的数据并将其减少到单个数据帧。
output.coalesce(1).write.format(\'json\').save(\'filtered.json\')它将输出以下数据：
{ “总”：328， “城市”： “内比都”}{ “总”：332， “城市”： “曼德勒”}{ “总”：340， “城市”： “仰光”}MySQL和Apache Spark集成
上述数据帧包含转换后的数据。我们希望将这些数据加载到MYSQL中，以便进一步使用，例如可视化或显示在应用程序上。
首先，我们需要MySQL连接器库来与Spark进行交互。我们将从MySQL网站下载连接器并将其放在一个文件夹中。我们将修改SparkSession以包含JAR文件。
scSpark = SparkSession \\ .builder \\ .appName("reading csv") \\ .config("spark.driver.extraClassPath", "/usr/local/spark/jars/mysql-connector-java-8.0.16.jar") \\ .getOrCreate()输出现在如下所示：
output = scSpark.sql(\'SELECT COUNT(*) as total, City from sales GROUP BY City\') output.show() output.write.format(\'jdbc\').options( url=\'jdbc:mysql://localhost/spark\', driver=\'com.mysql.cj.jdbc.Driver\', dbtable=\'city_info\', user=\'root\', password=\'root\').mode(\'append\').save()在运行脚本之前，我在数据库中创建了所需的Db和表。如果一切顺利，您应该看到如下结果：
如您所见，Spark可以更轻松地将数据从一个数据源传输到另一个数据源。
结论
Apache Spark是一个非常苛刻且有用的大数据工具，可以帮助您轻松编写ETL。 您可以加载pb级的数据，并且可以通过设置多个节点的集群轻松地处理这些数据。本教程只是为您提供Apache Spark编写ETL的基本思想。 您应该检查文档和其他资源以深入挖掘。
下载标准CJ/T 328-2010 球墨铸铁复合树脂水箅全文pdf文件。