本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 教學課程：與 Apache Spark 整合以匯入或匯出資料
Apache Spark 是用於大規模資料分析的開放原始碼引擎。Apache Spark 可讓您更有效率地對存放在 Amazon Keyspaces 中的資料進行分析。您也可以使用 Amazon Keyspaces 為應用程式提供一致的single-digit-millisecond讀取存取 Spark 的分析資料。開放原始碼 Spark Cassandra 連接器可簡化 Amazon Keyspaces 和 Spark 之間的讀取和寫入資料。
Amazon Keyspaces 對 Spark Cassandra Connector 的支援使用全受管無伺服器資料庫服務，簡化在 Spark 型分析管道中執行 Cassandra 工作負載的過程。使用 Amazon Keyspaces，您不需要擔心 Spark 與資料表競爭相同的基礎基礎設施資源。Amazon Keyspaces 資料表會根據您的應用程式流量自動擴展和縮減。
下列教學課程會逐步解說使用 Spark Cassandra 連接器將資料讀取和寫入 Amazon Keyspaces 所需的步驟和最佳實務。本教學課程示範如何使用 Spark Cassandra 連接器從檔案載入資料並將其寫入 Amazon Keyspaces 資料表，以將資料遷移至 Amazon Keyspaces。然後，本教學課程示範如何使用 Spark Cassandra 連接器從 Amazon Keyspaces 讀取資料。您會這麼做，以在 Spark 型分析管道中執行 Cassandra 工作負載。
**Topics**
+ [使用 Spark Cassandra 連接器建立 Amazon Keyspaces 連線的先決條件](spark-tutorial-prerequisites.md)
+ [步驟 1：設定 Amazon Keyspaces 以與 Apache Cassandra Spark Connector 整合](spark-tutorial-step1.md)
+ [步驟 2：設定 Apache Cassandra Spark 連接器](spark-tutorial-step2.md)
+ [步驟 3：建立應用程式組態檔案](spark-tutorial-step3.md)
+ [步驟 4：準備 Amazon Keyspaces 中的來源資料和目標資料表](spark-tutorial-step4.md)
+ [步驟 5：使用 Apache Cassandra Spark 連接器寫入和讀取 Amazon Keyspaces 資料](spark-tutorial-step5.md)
+ [對搭配 Amazon Keyspaces 使用 Spark Cassandra 連接器時的常見錯誤進行故障診斷](spark-tutorial-step6.md)
# 使用 Spark Cassandra 連接器建立 Amazon Keyspaces 連線的先決條件
使用 Spark Cassandra 連接器連線到 Amazon Keyspaces 之前，您需要確定已安裝下列項目。Amazon Keyspaces 與 Spark Cassandra Connector 的相容性已經過下列建議版本測試：
+ Java 第 8 版
+ Scala 2.12
+ Spark 3.4
+ Cassandra 連接器 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 許可。下列最佳實務可協助您為資料表佈建足夠的讀取/寫入容量。
1. 確認`Murmur3Partitioner`分割器是您帳戶的預設分割器。此分割區器與 Spark Cassandra 連接器相容。如需分割器以及如何變更分割器的詳細資訊，請參閱 [在 Amazon Keyspaces 中使用分割區](working-with-partitioners.md)。
1. 使用介面 VPC 端點搭配 Apache Spark 來設定 Amazon Keyspaces 的 IAM 許可。
+ 指派使用者資料表的讀取/寫入存取權，以及系統資料表的讀取存取權，如下列 IAM 政策範例所示。
+ 對於透過 VPC 端點透過 Spark 存取 Amazon Keyspaces 的用戶端，需要使用可用的介面 [VPC 端點](https://docs.aws.amazon.com/keyspaces/latest/devguide/vpc-endpoints.html)填入 system.peers 資料表。
```
{
"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 連接器的流量。
+ 開始使用隨需容量來協助您測試案例。
+ 若要最佳化生產環境的資料表輸送量成本，請針對連接器的流量使用速率限制器，並將資料表設定為使用佈建容量搭配自動擴展。如需詳細資訊，請參閱[使用 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 中設定讀取/寫入容量模式](ReadWriteCapacityMode.md)。
# 步驟 2：設定 Apache Cassandra Spark 連接器
Apache Spark 是以不同方式設定的通用運算平台。若要設定 Spark 和 Spark Cassandra Connector 以與 Amazon Keyspaces 整合，建議您從下一節所述的最低組態設定開始，然後在稍後根據您的工作負載進行增加。
+ **建立小於 8 MBs Spark 分割區大小。**
在 Spark 中，*分割區*代表可平行執行的資料原子區塊。當您使用 Spark Cassandra 連接器將資料寫入 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：建立應用程式組態檔案
若要搭配 Amazon Keyspaces 使用開放原始碼 Spark Cassandra Connector，您需要提供應用程式組態檔案，其中包含與 DataStax Java 驅動程式連線所需的設定。您可以使用服務特定的登入資料或 SigV4 外掛程式來連線。
如果您尚未這麼做，則需要將用來建立 TLS 連線的數位憑證轉換為 trustStore 檔案。您可以從 [開始之前](using_java_driver.md#using_java_driver.BeforeYouBegin) Java 驅動程式連線教學課程中遵循 的詳細步驟。請記下 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)。
在下列範例中，`password`使用您自己的登入資料取代 `username`和 。此外，請取代 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 連接器會為每個執行器部署 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. 建立來源檔案。您可以選擇下列其中一個選項：
+ 在本教學課程中，您會使用逗號分隔值 (CSV) 檔案，其名稱`keyspaces_sample_table.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. 使用 連線至 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 連接器寫入和讀取 Amazon Keyspaces 資料
在此步驟中，您會先`DataFrame`使用 Spark Cassandra 連接器，將範例檔案中的資料載入 。接著，您將資料從 寫入 `DataFrame` Amazon Keyspaces 資料表。您也可以獨立使用此部分，例如，將資料遷移至 Amazon Keyspaces 資料表。最後，您會`DataFrame`使用 Spark Cassandra 連接器，將資料表中的資料讀取至 。您也可以獨立使用此部分，例如，從 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 連接器。
```
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|
+--------------------+------------------+----------------+-----------+-------------+----+----+
```
# 對搭配 Amazon Keyspaces 使用 Spark Cassandra 連接器時的常見錯誤進行故障診斷
如果您使用 Amazon Virtual Private Cloud 並連線至 Amazon Keyspaces，則使用 Spark 連接器時最常見的錯誤是由下列組態問題所造成。
+ VPC 中使用的 IAM 使用者或角色缺少存取 Amazon Keyspaces 中`system.peers`資料表所需的許可。如需詳細資訊，請參閱[使用界面 VPC 端點資訊填入`system.peers`資料表項目](vpc-endpoints.md#system_peers)。
+ IAM 使用者或角色缺少使用者資料表的必要讀取/寫入許可，以及對 Amazon Keyspaces 中系統資料表的讀取存取權。如需詳細資訊，請參閱[步驟 1：設定 Amazon Keyspaces 以與 Apache Cassandra Spark Connector 整合](spark-tutorial-step1.md)。
+ Java 驅動程式組態不會在建立 SSL/TLS 連線時停用主機名稱驗證。如需範例，請參閱 [步驟 2：設定驅動程式](using_java_driver.md#java_tutorial.driverconfiguration)。
如需詳細的連線疑難排解步驟，請參閱 [我的 VPC 端點連線無法正常運作](troubleshooting.connecting.md#troubleshooting.connection.vpce)。
此外，您可以使用 Amazon CloudWatch 指標來協助您疑難排解 Amazon Keyspaces 中 Spark Cassandra Connector 組態的問題。若要進一步了解如何搭配 CloudWatch 使用 Amazon Keyspaces，請參閱 [使用 Amazon CloudWatch 監控 Amazon Keyspaces](monitoring-cloudwatch.md)。
下一節說明當您使用 Spark Cassandra 連接器時，應觀察的最有用指標。
**PerConnectionRequestRateExceeded**
Amazon Keyspaces 的配額為每個連線每秒 3，000 個請求。每個 Spark 執行器都會建立與 Amazon Keyspaces 的連線。執行多次重試可能會耗盡每個連線請求率配額。如果您超過此配額，Amazon Keyspaces 會在 CloudWatch 中發出`PerConnectionRequestRateExceeded`指標。
如果您看到 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)。您可以安全地忽略此警告。