本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 教程：与 Apache Spark 集成以导入或导出数据 Apache Spark 是一款用于大规模数据分析的开源引擎。Apache Spark 让您能够更有效地分析存储在 Amazon Keyspaces 中的数据。您还可以使用 Amazon Keyspaces 为应用程序提供对 Spark 分析数据的一致 single-digit-millisecond读取权限。开源 Spark Cassandra Connector 可以简化 Amazon Keyspaces 和 Spark 之间的数据读写。 Amazon Keyspaces 可以使用完全托管的无服务器数据库服务来简化 Cassandra 工作负载在基于 Spark 的分析管道中的运行，从而对 Spark Cassandra Connector 提供支持。有了 Amazon Keyspaces，您无需担心 Spark 会与您的表争夺底层基础设施资源。Amazon Keyspaces 表会根据您的应用程序流量自动扩缩。 以下教程将会介绍使用 Spark Cassandra Connector 向 Amazon Keyspaces 读取和写入数据所需的步骤和最佳实践。本教程演示了如何使用 Spark Cassandra Connector 从文件中加载数据并将其写入 Amazon Keyspaces 表，从而将数据迁移到 Amazon Keyspaces。然后，本教程展示了如何使用 Spark Cassandra Connector 从 Amazon Keyspaces 读回数据。进行这一操作的目的是在基于 Spark 的分析管道中运行 Cassandra 工作负载。 **Topics** + [使用 Spark Cassandra Connector 建立与 Amazon Keyspaces 的连接的先决条件](spark-tutorial-prerequisites.md) + [第 1 步：配置 Amazon Keyspaces 以便与 Apache Cassandra Spark Connector 集成](spark-tutorial-step1.md) + [步骤 2：配置 Apache Cassandra Spark Connector](spark-tutorial-step2.md) + [步骤 3：创建应用程序配置文件](spark-tutorial-step3.md) + [步骤 4：在 Amazon Keyspaces 中准备源数据和目标表](spark-tutorial-step4.md) + [步骤 5：使用 Apache Cassandra Spark Connector 写入和读取 Amazon Keyspaces 数据](spark-tutorial-step5.md) + [排除将 Spark Cassandra Connector 与 Amazon Keyspaces 配合使用时的常见错误](spark-tutorial-step6.md) # 使用 Spark Cassandra Connector 建立与 Amazon Keyspaces 的连接的先决条件 在使用 Spark Cassandra Connector 连接到 Amazon Keyspaces 之前，您需要确保安装以下内容。Amazon Keyspaces 与 Spark Cassandra Connector 的兼容性已通过以下推荐版本进行了测试： + Java 版本 8 + Scala 2.12 + Spark 3.4 + Cassandra Connector 2.5 及更高版本 + Cassandra 驱动程序 4.12 1. 要安装 Scala，请按照 [https://www.scala-lang.org/download/scala2.html](https://www.scala-lang.org/download/scala2.html) 中的说明进行操作。 1. 要安装 Spark 3.4.1，请按照以下示例进行操作。 ``` curl -o spark-3.4.1-bin-hadoop3.tgz -k https://dlcdn.apache.org/spark/spark-3.4.1/spark-3.4.1-bin-hadoop3.tgz # now to untar tar -zxvf spark-3.4.1-bin-hadoop3.tgz # set this variable. export SPARK_HOME=$PWD/spark-3.4.1-bin-hadoop3 ``` ``` # 第 1 步：配置 Amazon Keyspaces 以便与 Apache Cassandra Spark Connector 集成 在本步骤中，您将确认您的账户的分区程序是否与 Apache Spark Connector 兼容，并设置所需的 IAM 权限。以下最佳实践可帮助您为表配置足够的 read/write 容量。 1. 确认 `Murmur3Partitioner` 分区程序是您账户的默认分区程序。该分区程序与 Spark Cassandra Connector 兼容。有关分区程序以及如何更改分区程序的更多信息，请参阅[使用 Amazon Keyspaces 中的分区程序](working-with-partitioners.md)。 1. 使用 Apache Spark 和接口 VPC 端点为 Amazon Keyspaces 设置 IAM 权限。 + 分配对用户表的 read/write 访问权限和对系统表的读取权限，如下面列出的 IAM 策略示例所示。 + 使用 Spark 通过 VPC 端点访问 Amazon Keyspaces 的客户端需要在 system.peers 表中填充您的可用接口 [VPC 端点](https://docs.aws.amazon.com/keyspaces/latest/devguide/vpc-endpoints.html)。 ``` { "Version":"2012-10-17", "Statement":[ { "Effect":"Allow", "Action":[ "cassandra:Select", "cassandra:Modify" ], "Resource":[ "arn:aws:cassandra:us-east-1:111122223333:/keyspace/mykeyspace/table/mytable", "arn:aws:cassandra:us-east-1:111122223333:/keyspace/system*" ] }, { "Sid":"ListVPCEndpoints", "Effect":"Allow", "Action":[ "ec2:DescribeNetworkInterfaces", "ec2:DescribeVpcEndpoints" ], "Resource":"*" } ] } ``` 1. 请考虑以下最佳实践，为您的 Amazon Keyspaces 表配置足够的读/写吞吐容量，以支持来自 Spark Cassandra Connector 的流量。 + 开始使用按需容量来帮助您测试场景。 + 要优化生产环境的表吞吐量成本，请对来自 Connector 的流量使用速率限制器，并将您的表配置为使用具有自动扩缩功能的预配置容量。有关更多信息，请参阅 [使用 Amazon Keyspaces 自动扩缩自动管理吞吐能力](autoscaling.md)。 + 你可以使用 Cassandra 驱动程序附带的固定速率限制器。[AWS 示例](https://github.com/aws-samples)存储库中有一些[为 Amazon Keyspaces 量身定制的速率限制器](https://github.com/aws-samples/amazon-keyspaces-java-driver-helpers)。 + 有关容量管理的更多信息，请参阅[在 Amazon Keyspaces 中配置 read/write 容量模式](ReadWriteCapacityMode.md)。 # 步骤 2：配置 Apache Cassandra Spark Connector Apache Spark 是一种通用计算平台，可以用不同的方式进行配置。要配置 Spark 和 Spark Cassandra Connector 以便与 Amazon Keyspaces 集成，我们建议您从下一节介绍的最低配置设置开始，然后根据您的工作负载提高配置设置。 + **创建大小小于 8 的 Spark 分区 MBs。** 在 Spark 中，*分区*代表可以并行运行的原子数据块。当您使用 Spark Cassandra Connector 向 Amazon Keyspaces 写入数据时，Spark 分区越小，任务要写入的记录量就越小。如果 Spark 任务遇到多个错误，则在进行指定次数的重试后，任务将会失败。为避免重新执行大型任务和重新处理大量数据，请将 Spark 分区的大小保持在较小的范围内。 + **每个执行程序采用较低的并发写入次数和较高的重试次数。** 如果操作超时，Amazon Keyspaces 会向 Cassandra 驱动程序返回容量不足错误。您无法通过更改配置的超时持续时间来解决因容量不足而导致的超时问题，因为 Spark Cassandra Connector 会尝试使用 `MultipleRetryPolicy` 以透明方式重试请求。为确保重试不会让驱动程序的连接池不堪重负，请让每个执行程序采用较低的并发写入次数和较高的重试次数。以下代码段是一个这方面的示例。 ``` spark.cassandra.query.retry.count = 500 spark.cassandra.output.concurrent.writes = 3 ``` + **分解总吞吐量并将其分配到多个 Cassandra 会话中。** + Cassandra Spark Connector 会为每个 Spark 执行程序创建一个会话。我们可以把这一会话视为确定所需吞吐量和所需连接数量的扩展单元。 + 在定义每个执行程序的内核数和每个任务的内核数时，请从低开始并根据需要增加数量。 + 设置 Spark 任务失败次数以便在出现暂时错误时继续处理。在您熟悉应用程序的流量特征和要求后，我们建议将 `spark.task.maxFailures` 设置为一个有界值。 + 例如，以下配置可以让每个执行程序在每个会话中处理两个并发任务： ``` spark.executor.instances = configurable -> number of executors for the session. spark.executor.cores = 2 -> Number of cores per executor. spark.task.cpus = 1 -> Number of cores per task. spark.task.maxFailures = -1 ``` + **关闭批处理。** + 我们建议您关闭批处理以改进随机访问模式。以下代码段是一个这方面的示例。 ``` spark.cassandra.output.batch.size.rows = 1 (Default = None) spark.cassandra.output.batch.grouping.key = none (Default = Partition) spark.cassandra.output.batch.grouping.buffer.size = 100 (Default = 1000) ``` + **将 `SPARK_LOCAL_DIRS` 设置为具有足够空间的本地高速磁盘。** + 默认情况下，Spark 会将地图输出文件和弹性分布式数据集 (RDDs) 保存到`/tmp `文件夹。根据您的 Spark 主机的配置，这可能会导致*设备没有剩余空间*这一类型的错误。 + 要将 `SPARK_LOCAL_DIRS` 环境变量设置为名为 `/example/spark-dir` 的目录，您可以使用以下命令。 ``` export SPARK_LOCAL_DIRS=/example/spark-dir ``` # 步骤 3：创建应用程序配置文件 要将开源 Spark Cassandra Connector 与 Amazon Keyspaces 配合使用，您需要提供一个包含连接 Java 驱动程序所需的设置 DataStax 的应用程序配置文件。您可以使用特定于服务的凭证或 SigV4 插件进行连接。 如果您尚未这样做，则需要将用于创建 TLS 连接的数字证书转换为 TrustStore 文件。您可以按照 Java 驱动程序连接教程的[开始前的准备工作](using_java_driver.md#using_java_driver.BeforeYouBegin)中的详细步骤进行操作。请记下 trustStore 文件路径和密码，因为您在创建应用程序配置文件时需要使用这一信息。 ## 使用 Sigv4 身份验证进行连接 本节向您展示了一个示例`application.conf`文件，您可以在连接 AWS 凭据和 SigV4 插件时使用该文件。如果您尚未这样做，则需要生成 IAM 访问密钥（访问密钥 ID 和私有访问密钥），并将其保存在 AWS 配置文件中或作为环境变量保存。有关详细说明，请参阅[适用于 Cassandra AWS CLI客户端驱动程序的、 AWS 软件开发工具包或 Amazon Keyspaces Sigv4 插件所需的凭证](SigV4_credentials.md)。 在以下示例中，请将文件路径替换为您的 trustStore 文件，并且替换密码。 ``` datastax-java-driver { basic.contact-points = ["cassandra.us-east-1.amazonaws.com:9142"] basic.load-balancing-policy { class = DefaultLoadBalancingPolicy local-datacenter = us-east-1 slow-replica-avoidance = false } basic.request { consistency = LOCAL_QUORUM } advanced { auth-provider = { class = software.aws.mcs.auth.SigV4AuthProvider aws-region = us-east-1 } ssl-engine-factory { class = DefaultSslEngineFactory truststore-path = "path_to_file/cassandra_truststore.jks" truststore-password = "password" hostname-validation=false } } advanced.connection.pool.local.size = 3 } ``` 更新此配置文件并将其保存为 `/home/user1/application.conf`。下列示例使用这一路径。 ## 使用特定于服务的凭证进行连接 本节向您展示了一个示例 `application.conf` 文件，您可以在使用特定于服务的凭证进行连接时使用这一文件。如果您尚未执行这一操作，则需要为 Amazon Keyspaces 生成特定于服务的凭证。有关详细说明，请参阅[创建用于通过编程方式访问 Amazon Keyspaces 的服务特定凭证。](programmatic.credentials.ssc.md)。 在以下示例中，将 `username` 和 `password` 替换为您自己的凭证。另外，请将文件路径替换为您的 trustStore 文件，并且替换密码。 ``` datastax-java-driver { basic.contact-points = ["cassandra.us-east-1.amazonaws.com:9142"] basic.load-balancing-policy { class = DefaultLoadBalancingPolicy local-datacenter = us-east-1 } basic.request { consistency = LOCAL_QUORUM } advanced { auth-provider = { class = PlainTextAuthProvider username = "username" password = "password" aws-region = "us-east-1" } ssl-engine-factory { class = DefaultSslEngineFactory truststore-path = "path_to_file/cassandra_truststore.jks" truststore-password = "password" hostname-validation=false } metadata = { schema { token-map.enabled = true } } } } ``` 更新此配置文件并将其保存为 `/home/user1/application.conf`，以便与代码示例一起使用。 ## 设置固定连接速率 要强制每个 Spark 执行程序都使用固定速率，您可以定义一个请求调节器。请求调节器可以限制每秒的请求速率。Spark Cassandra Connector 会为每个执行程序都部署一个 Cassandra 会话。使用以下公式可以帮助您在表中实现一致的吞吐量。 ``` max-request-per-second * numberOfExecutors = total throughput against a table ``` 您可以将此示例添加到之前创建的应用程序配置文件中。 ``` datastax-java-driver { advanced.throttler { class = RateLimitingRequestThrottler max-requests-per-second = 3000 max-queue-size = 30000 drain-interval = 1 millisecond } } ``` # 步骤 4：在 Amazon Keyspaces 中准备源数据和目标表 在本步骤中，您将创建一个包含示例数据和 Amazon Keyspaces 表的源文件。 1. 创建源文件。您可以选择以下选项之一： + 在本教程中，您会使用名为 `keyspaces_sample_table.csv` 的逗号分隔值 (CSV) 文件作为数据迁移的源文件。提供的示例文件包含名为 `book_awards` 的表中的几行数据。 1. 下载以下存档文件 [samplemigration.zip](samples/samplemigration.zip) 中包含的示例 CSV 文件 (`keyspaces_sample_table.csv`)。解压缩存档文件并记下指向 `keyspaces_sample_table.csv` 的路径。 + 如果您想使用自己的 CSV 文件将数据写入 Amazon Keyspaces，请确保数据经过随机化处理。直接从数据库读取或导出到平面文件的数据通常按分区和主键排序。将有序的数据导入 Amazon Keyspaces 可能会导致数据被写入较小的 Amazon Keyspaces 分区，从而造成流量分布不均匀。这可能会导致性能降低和错误率上升。 相比之下，将数据随机化有助于更均匀地在分区之间分配流量，从而利用 Amazon Keyspaces 内置的负载均衡功能。您可以使用多种工具将数据随机化。有关使用开源工具 [Shuf](https://en.wikipedia.org/wiki/Shuf) 的示例，请参阅数据迁移教程中的[第 2 步：使用准备要上传的数据 DSBulk](dsbulk-upload-prepare-data.md)。以下示例展示了如何将数据随机处理为 `DataFrame`。 ``` import org.apache.spark.sql.functions.randval shuffledDF = dataframe.orderBy(rand()) ``` 1. 在 Amazon Keyspaces 中创建目标键空间和表。 1. 使用 Connect 到 Amazon Keyspaces。`cqlsh-expansion`有关 `cqlsh-expansion` 安装说明，请参阅[使用 `cqlsh-expansion` 连接 Amazon Keyspaces](programmatic.cqlsh.md#using_cqlsh)。 将以下示例中的服务端点替换为您自己的值。 ``` cqlsh-expansion cassandra.us-east-1.amazonaws.com 9142 --ssl ``` 1. 使用以下示例中所示的名称 `catalog` 创建新的键空间。 ``` CREATE KEYSPACE catalog WITH REPLICATION = {'class': 'SingleRegionStrategy'}; ``` 1. 在新键空间变为可用状态后，使用以下代码创建目标表 `book_awards`。要了解有关异步资源创建以及如何检查资源是否可用的更多信息，请参阅[在 Amazon Keyspaces 中查看键空间的创建状态](keyspaces-create.md)。 ``` CREATE TABLE catalog.book_awards ( year int, award text, rank int, category text, book_title text, author text, publisher text, PRIMARY KEY ((year, award), category, rank) ); ``` # 步骤 5：使用 Apache Cassandra Spark Connector 写入和读取 Amazon Keyspaces 数据 在本步骤中，您首先需要使用 Spark Cassandra Connector 将示例文件中的数据加载到 `DataFrame` 中。接下来，您需要将数据从 `DataFrame` 写入您的 Amazon Keyspaces 表中。您也可以单独使用这一部分执行其他操作，比如将数据迁移到 Amazon Keyspaces 表中。最后，您需要使用 Spark Cassandra Connector 将表中的数据读入到一个 `DataFrame` 中。您也可以单独使用这一部分执行其他操作，比如从 Amazon Keyspaces 表中读取数据，以便使用 Apache Spark 执行数据分析。 1. 启动 Spark Shell，如以下示例所示。请注意，此示例使用的是 Sigv4 身份验证。 ``` ./spark-shell --files application.conf --conf spark.cassandra.connection.config.profile.path=application.conf --packages software.aws.mcs:aws-sigv4-auth-cassandra-java-driver-plugin:4.0.5,com.datastax.spark:spark-cassandra-connector_2.12:3.1.0 --conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions ``` 1. 使用以下代码导入 Spark Cassandra Connector。 ``` import org.apache.spark.sql.cassandra._ ``` 1. 要从 CSV 文件中读取数据并将其存储在 `DataFrame` 中，您可以使用以下示例代码。 ``` var df = spark.read.option("header","true").option("inferSchema","true").csv("keyspaces_sample_table.csv") ``` 您可以使用以下命令来显示结果。 ``` scala> df.show(); ``` 输出应如下所示： ``` +----------------+----+-----------+----+------------------+--------------------+-------------+ | award|year| category|rank| author| book_title| publisher| +----------------+----+-----------+----+------------------+--------------------+-------------+ |Kwesi Manu Prize|2020| Fiction| 1| Akua Mansa| Where did you go?|SomePublisher| |Kwesi Manu Prize|2020| Fiction| 2| John Stiles| Yesterday|Example Books| |Kwesi Manu Prize|2020| Fiction| 3| Nikki Wolf|Moving to the Cha...| AnyPublisher| | Wolf|2020|Non-Fiction| 1| Wang Xiulan| History of Ideas|Example Books| | Wolf|2020|Non-Fiction| 2|Ana Carolina Silva| Science Today|SomePublisher| | Wolf|2020|Non-Fiction| 3| Shirley Rodriguez|The Future of Sea...| AnyPublisher| | Richard Roe|2020| Fiction| 1| Alejandro Rosalez| Long Summer|SomePublisher| | Richard Roe|2020| Fiction| 2| Arnav Desai| The Key|Example Books| | Richard Roe|2020| Fiction| 3| Mateo Jackson| Inside the Whale| AnyPublisher| +----------------+----+-----------+----+------------------+--------------------+-------------+ ``` 您可以确认 `DataFrame` 中的数据的架构，如以下示例所示。 ``` scala> df.printSchema ``` 输出应如下所示： ``` root |-- award: string (nullable = true) |-- year: integer (nullable = true) |-- category: string (nullable = true) |-- rank: integer (nullable = true) |-- author: string (nullable = true) |-- book_title: string (nullable = true) |-- publisher: string (nullable = true) ``` 1. 使用以下命令将 `DataFrame` 中的数据写入 Amazon Keyspaces 表中。 ``` df.write.cassandraFormat("book_awards", "catalog").mode("APPEND").save() ``` 1. 要确认数据已保存，您可以将其读回到一个数据框，如以下示例所示。 ``` var newDf = spark.read.cassandraFormat("book_awards", "catalog").load() ``` 然后，你可以显示现在包含在数据框中的数据。 ``` scala> newDf.show() ``` 该命令应生成如下所示的输出。 ``` +--------------------+------------------+----------------+-----------+-------------+----+----+ | book_title| author| award| category| publisher|rank|year| +--------------------+------------------+----------------+-----------+-------------+----+----+ | Long Summer| Alejandro Rosalez| Richard Roe| Fiction|SomePublisher| 1|2020| | History of Ideas| Wang Xiulan| Wolf|Non-Fiction|Example Books| 1|2020| | Where did you go?| Akua Mansa|Kwesi Manu Prize| Fiction|SomePublisher| 1|2020| | Inside the Whale| Mateo Jackson| Richard Roe| Fiction| AnyPublisher| 3|2020| | Yesterday| John Stiles|Kwesi Manu Prize| Fiction|Example Books| 2|2020| |Moving to the Cha...| Nikki Wolf|Kwesi Manu Prize| Fiction| AnyPublisher| 3|2020| |The Future of Sea...| Shirley Rodriguez| Wolf|Non-Fiction| AnyPublisher| 3|2020| | Science Today|Ana Carolina Silva| Wolf|Non-Fiction|SomePublisher| 2|2020| | The Key| Arnav Desai| Richard Roe| Fiction|Example Books| 2|2020| +--------------------+------------------+----------------+-----------+-------------+----+----+ ``` # 排除将 Spark Cassandra Connector 与 Amazon Keyspaces 配合使用时的常见错误 如果您在使用 Amazon Virtual Private Cloud 并连接到 Amazon Keyspaces，则在使用 Spark 连接器时遇到的最常见的错误往往由以下配置问题引起。 + VPC 中使用的 IAM 用户或角色缺少访问 Amazon Keyspaces 中的 `system.peers` 表所需的权限。有关更多信息，请参阅 [使用接口 VPC 端点信息填充 `system.peers` 表条目](vpc-endpoints.md#system_peers)。 + IAM 用户或角色缺乏所需的用户表 read/write 权限，也缺乏对 Amazon Keyspaces 中系统表的读取权限。有关更多信息，请参阅 [第 1 步：配置 Amazon Keyspaces 以便与 Apache Cassandra Spark Connector 集成](spark-tutorial-step1.md)。 + 创建 SSL/TLS 连接时，Java 驱动程序配置没有禁用主机名验证。有关示例，请参阅 [步骤 2：配置驱动程序](using_java_driver.md#java_tutorial.driverconfiguration)。 有关详细的链接故障排除步骤，请参阅[我的 VPC 端点连接无法正常工作](troubleshooting.connecting.md#troubleshooting.connection.vpce)。 此外，您还可以使用亚马逊 CloudWatch 指标来帮助您解决亚马逊密钥空间中 Spark Cassandra Connector 配置的问题。要了解有关将 Amazon Keyspaces 与配合使用的更多信息 CloudWatch，请参阅。[使用亚马逊监控亚马逊密钥空间 CloudWatch](monitoring-cloudwatch.md) 以下部分介绍了在使用 Spark Cassandra Connector 时需要观察的最有用的指标。 **PerConnectionRequestRateExceeded** Amazon Keyspaces 的配额为每个连接每秒 3000 个请求。每个 Spark 执行程序都会与 Amazon Keyspaces 建立连接。多次重试可能会耗尽每个连接的请求速率配额。如果您超过此配额，Amazon Keyspaces 将发布一个`PerConnectionRequestRateExceeded`指标。 CloudWatch 如果您看到 PerConnectionRequestRateExceeded 事件以及其他系统或用户错误，则可能是 Spark 正在运行多次重试，超出每个连接分配的请求数。 如果您看到 `PerConnectionRequestRateExceeded` 事件但没有看到其他错误，则您可能需要增加驱动程序设置中的连接数以提高吞吐量，或者可能需要增加 Spark 作业中的执行程序数量。 **StoragePartitionThroughputCapacityExceeded** Amazon Keyspaces 的每个分区的配额为 1,000 WCUs 或 WRUs 每秒/3,000 RCUs 或 RRUs 每秒。如果您看到`StoragePartitionThroughputCapacityExceeded` CloudWatch 事件，则可能表明数据在加载时没有随机化。有关如何随机处理数据的示例，请参阅[步骤 4：在 Amazon Keyspaces 中准备源数据和目标表](spark-tutorial-step4.md)。 ## 常见错误和警告 如果您在使用 Amazon Virtual Private Cloud 并连接到 Amazon Keyspaces，则 Cassandra 驱动程序可能会在 `system.peers` 表中发出有关控制节点本身的警告消息。有关更多信息，请参阅 [常见错误和警告](vpc-endpoints.md#vpc_troubleshooting)。您可以放心地忽略这一警告。