Spark分布式技巧：数据库读取指南 (spark怎么分布式读取数据库)

Introduction

随着大数据和云计算的兴起，分布式计算框架成为数据处理的主流。Apache Spark是一个用于大规模数据处理的强大分布式计算框架。它可以在分布式集群上快速地处理数据，并提供操作Hadoop Distributed File System（HDFS）和其他数据源的功能。在这篇文章中，我们将讨论在Spark中使用数据库读取技巧的指南。我们将深入了解Spark如何处理数据库中的数据，优化代码性能以及使用Spark SQL等工具来提高数据处理过程的效率。

Spark中的数据库读取技巧

Spark提供了各种方法来读取和处理数据库中的数据。下面是一些我们需要了解的技巧。

1. JDBC连接

Spark可以使用Java Database Connectivity（JDBC）来连接关系型数据库。JDBC是一种Java API，用于与关系型数据库建立连接。Spark可以通过JDBC读取数据并进行转换。下面是一个基本的连接示例：

“`

val jdbcDF = spark.read

.format(“jdbc”)

.option(“url”, “jdbc:postgresql://localhost:5432/mydatabase”)

.option(“dbtable”, “mytable”)

.option(“user”, “myuser”)

.option(“password”, “mypassword”)

.load()

“`

在这种情况下，我们使用“postgres”数据库的JDBC驱动程序连接到本地端口5432上的“mydatabase”数据库。然后，我们将“mytable”表加载到数据框架中。我们还需要提供用户名和密码来连接到数据库。

2. 数据分区

为了获得更好的性能，我们应该将数据分割为多个分区，然后在集群上并行处理。可通过以下代码指定分区数：

“`

val jdbcDF = spark.read

.format(“jdbc”)

.option(“url”, “jdbc:postgresql://localhost:5432/mydatabase”)

.option(“dbtable”, “mytable”)

.option(“user”, “myuser”)

.option(“password”, “mypassword”)

.option(“partitionColumn”, “id”)

.option(“lowerBound”, “1”)

.option(“upperBound”, “100000”)

.option(“numPartitions”, “16”)

.load()

“`

在这个例子中，我们在“id”列中使用分区，它的最小值为1，更大值为100000，总共有16个分区。

3. 自定义查询

从数据库中选择大量数据可能会导致Spark出现内存问题。如果我们只需要部分列或部分行数据，则可以使用自定义查询。我们可以通过以下代码将自定义查询添加到我们的Spark应用程序中：

“`

val query = “(SELECT name, age FROM mytable WHERE age > 20) as myquery”

val jdbcDF = spark.read

.format(“jdbc”)

.option(“url”, “jdbc:postgresql://localhost:5432/mydatabase”)

.option(“dbtable”, query)

.option(“user”, “myuser”)

.option(“password”, “mypassword”)

.load()

“`

请注意，我们现在查询的不是完整的表，而是只查询名字和年龄大于20岁的行。

4. 使用Spark SQL

Spark SQL提供了一个快速和方便的方式来处理和查询数据库数据。可以通过以下代码使用Spark SQL读取并查询数据库数据：

“`

val jdbcDF = spark.read

.format(“jdbc”)

.option(“url”, “jdbc:postgresql://localhost:5432/mydatabase”)

.option(“dbtable”, “mytable”)

.option(“user”, “myuser”)

.option(“password”, “mypassword”)

.load()

jdbcDF.createOrReplaceTempView(“mytable”)

val result = spark.sql(“SELECT * FROM mytable WHERE age > 20”)

“`

在这个例子中，我们将数据库数据加载到数据框架中，并使用createOrReplaceTempView（）将它们转换为Spark SQL表格。然后，我们可以使用SQL语句来查询这些数据。

优化技巧

在处理大量数据库数据时，性能是一个关键问题。以下是一些优化技巧，以提高处理速度和效率。

1. 分区和缓存

Spark将我们的数据分成分区，以便可以在集群上并行运行操作。如果我们查询的数据集很大，我们应该将数据缓存到内存中，以避免重复加载和处理。可以使用以下代码将DataFrame缓存到内存中：

“`

jdbcDF.persist(StorageLevel.MEMORY_ON)

“`

2. 数据类型

Spark需要知道每个数据列的类型。如果Spark不知道一个列是什么类型，它将使用字符串类型，并且内存使用率会增加。可以使用以下代码来指定每个列的数据类型：

“`

val schema = StructType(Array(

StructField(“id”, IntegerType, true),

StructField(“name”, StringType, true),

StructField(“age”, IntegerType, true))

)

val jdbcDF = spark.read

.format(“jdbc”)

.option(“url”, “jdbc:postgresql://localhost:5432/mydatabase”)

.option(“dbtable”, “mytable”)

.option(“user”, “myuser”)

.option(“password”, “mypassword”)

.schema(schema)

.load()

“`

在这个例子中，我们使用schema（）方法指定每个列的数据类型。

3. 缓存和Table

Spark SQL中的缓存和表也可以提高性能。类似于数据框架的缓存，我们可以使用以下代码将结果缓存到内存中：

“`

result.cache()

“`

我们还可以使用以下代码将结果保存到Spark SQL表中：

“`

result.write.format(“parquet”).saveAsTable(“myresult”)

“`

这将使我们能够在我们的应用程序中随时查询数据，并避免重复计算。

结论

Spark是一个强大的分布式计算框架，可以处理大量数据源。在本文中，我们已讨论如何使用JDBC连接数据库、数据分区、自定义查询和Spark SQL等技术，优化性能以及使用Spark SQL进行查询。通过这些技术和优化，我们可以更快地处理数据库数据，并在大规模使用中提高我们的性能和效率。

相关问题拓展阅读：

• spark1.2.1实现读取hbase的数据后怎么实现实时查询
• Hadoop与分布式数据处理 Spark VS Hadoop有哪些异同点

spark1.2.1实现读取hbase的数据后怎么实现实时查询

调用parallelize函数直接从中获取数据，并存入RDD中；Java版本如下：

JavaRDD myRDD = sc.parallelize(Arrays.asList(1,2,3));

　　Scala版本如下：

val myRDD= sc.parallelize(List(1,2,3))

Hadoop与分布式数据处理 Spark VS Hadoop有哪些异同点

Hadoop

分布式批处理计算，强调批处理，常用于数据挖掘、分析

   Spark 

是一个基于内存计算的开源的集群计算系统，目的是让数据分析更加快速, Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。

      Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地对象一样轻松地操作分布式数据集。

     尽管创建 Spark 是为了支持分布式数据集上的迭代作业，但是实际上它是对 Hadoop 的补充，可以在 Hadoop 文件系统中并行运行。通过名为Mesos的第三方集群框架可以支持此行为。Spark 由加州大学伯克利分校 AMP 实验室 (Algorithms,Machines,and People Lab) 开发，可用来构建大型的、低延迟的数据分析应用程序。

      虽然 Spark 与 Hadoop 有相似之处，但它提供了具有圆数笑有用差异的一个新的集群计算框架。首先，Spark 是为集群计算中的特定类型的工作负载而设计，即那些在并行操作之间重用工作数据集（比如机器学习算法）橘含的工作负载。为了优化这些类型的工作负载，Spark 引进了内存集群计算的概念，可在毕唤内存集群计算中将数据集缓存在内存中，以缩短访问延迟.

1、解决问题的层面不一样

首先，Hadoop和Apache Spark两者都是大数据框架，但是各自存在的目的不尽相同。Hadoop实质信含上更多是一个分布式数据基础设施: 它将巨大的数据集分派到一个由普通计算机组成的集群中的多个节点进行存储，意味着您不需要购买和维护昂贵的服务器硬件。

同时，Hadoop还会索引和跟踪这些数据，让大数据处理和分析效率达到前所未有的高度。Spark，则是那么一个专门用来对那些分布式存储的大数据进行处理的工具，它并不会进行分布式数据的存储。

2、两者可合可分

Hadoop除了提供为大家所共识的HDFS分布式数据存储功能之外，还提供了叫做MapReduce的数据处理功能。所以这里我们完全可以抛开Spark，使用Hadoop自身的MapReduce来完成数据的处理。

相反，Spark也不是非要依附在Hadoop身上才能生存。但如上所述，毕竟它没有提供文件管理系统，所以，它必须和桐洞其他的分布式文件系统进行集成才能运作。这里我们可以选择Hadoop的HDFS,也可以选择其他的基于云的数据系统平台。但Spark默认来说还是被用在Hadoop上面的，毕竟，大家都认为它们的结合是更好的。

以下是从网上摘录的对MapReduce的最简洁明了的解析：

我们要数图书馆中的所有书。你数1号书架，我数2号书架。这就是“Map”。我们人越多，数书就更快。

现在我们到一起，把所有人的统计数加在一起。这就是“Reduce”。

3、Spark数据处理速度秒杀MapReduce

Spark因为其处理数据的方式不一样，会比MapReduce快上很多。MapReduce是分步对数据进行处理的: ”从集群中读取数据，进行一次处理，将结果写到集群，从集群中读取更新后的数据，进行下一次的处理，将结果写到集群，等等…“ Booz Allen Hamilton的数据科学家Kirk Borne如此解析。

反观Spark，它会在内存中以接近“实时”的时间完成所有的数据分析：“从集群中读取数据，完成所有必须的分析处理，将结果写回集群，完成，” Born说道。Spark的批处理速度比MapReduce快近10倍，内存中的数据分析速度则快近100倍。

如果需要处理的数据和结果需求大部分情况下是静态的，且你也有耐心等待批处理的完成的话，MapReduce的处理方式也是完全可以接受的。

但如果你需要对流数据进行分析，比如那些来自于工厂的传感器收集回来的数据，又或者说你的应用是需要多重数据处理的，那么你也许更应滑轮笑该使用Spark进行处理。

大部分机器学习算法都是需要多重数据处理的。此外，通常会用到Spark的应用场景有以下方面：实时的市场活动，在线产品推荐，网络安全分析，机器日记监控等。

4、灾难恢复

两者的灾难恢复方式迥异，但是都很不错。因为Hadoop将每次处理后的数据都写入到磁盘上，所以其天生就能很有弹性的对系统错误进行处理。

Spark的数据对象存储在分布于数据集群中的叫做弹性分布式数据集(RDD: Resilient Distributed Dataset)中。这些数据对象既可以放在内存，也可以放在磁盘，所以RDD同样也可以提供完成的灾难恢复功能。

关于spark怎么分布式读取数据库的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。

香港服务器首选树叶云，2H2G首月10元开通。
树叶云（shuyeidc.com）提供简单好用，价格厚道的香港/美国云服务器和独立服务器。IDC+ISP+ICP资质。ARIN和APNIC会员。成熟技术团队15年行业经验。