Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
# Mit Amazon EMR Serverless eine Verbindung zu DynamoDB herstellen
In diesem Tutorial laden Sie eine Teilmenge der Daten vom [Vereinigte Staaten Board on Geographic Names](https://www.usgs.gov/us-board-on-geographic-names) in einen Amazon S3 S3-Bucket hoch und kopieren die Daten dann mit Hive oder Spark auf Amazon EMR Serverless zur Abfrage in eine Amazon DynamoDB-Tabelle.
## Schritt 1: Daten in einen Amazon S3 S3-Bucket hochladen
Um einen Amazon S3 S3-Bucket zu erstellen, folgen Sie den Anweisungen [unter Bucket erstellen](https://docs.aws.amazon.com/AmazonS3/latest/user-guide/create-bucket.html) im *Amazon Simple Storage Service Console-Benutzerhandbuch*. Ersetzen Sie Verweise auf `amzn-s3-demo-bucket` durch den Namen Ihres neu erstellten Buckets. Jetzt ist Ihre EMR Serverless-Anwendung bereit, Jobs auszuführen.
1. Laden Sie das Beispieldatenarchiv `features.zip` mit dem folgenden Befehl herunter.
```
wget https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/samples/features.zip
```
1. Extrahieren Sie die `features.txt` Datei aus dem Archiv und greifen Sie auf die ersten Zeilen der Datei zu:
```
unzip features.zip
head features.txt
```
Das Ergebnis sollte dem Folgenden ähneln.
```
1535908|Big Run|Stream|WV|38.6370428|-80.8595469|794
875609|Constable Hook|Cape|NJ|40.657881|-74.0990309|7
1217998|Gooseberry Island|Island|RI|41.4534361|-71.3253284|10
26603|Boone Moore Spring|Spring|AZ|34.0895692|-111.410065|3681
1506738|Missouri Flat|Flat|WA|46.7634987|-117.0346113|2605
1181348|Minnow Run|Stream|PA|40.0820178|-79.3800349|1558
1288759|Hunting Creek|Stream|TN|36.343969|-83.8029682|1024
533060|Big Charles Bayou|Bay|LA|29.6046517|-91.9828654|0
829689|Greenwood Creek|Stream|NE|41.596086|-103.0499296|3671
541692|Button Willow Island|Island|LA|31.9579389|-93.0648847|98
```
Die Felder in jeder Zeile geben hier eine eindeutige Kennung, einen Namen, die Art des natürlichen Merkmals, den Bundesstaat, den Breitengrad in Grad, den Längengrad in Grad und die Höhe in Fuß an.
1. Laden Sie Ihre Daten auf Amazon S3 hoch
```
aws s3 cp features.txt s3://amzn-s3-demo-bucket/features/
```
## Schritt 2: Erstellen Sie eine Hive-Tabelle
Verwenden Sie Apache Spark oder Hive, um eine neue Hive-Tabelle zu erstellen, die die hochgeladenen Daten in Amazon S3 enthält.
------
#### [ Spark ]
Um eine Hive-Tabelle mit Spark zu erstellen, führen Sie den folgenden Befehl aus.
```
import org.apache.spark.sql.SparkSession
val sparkSession = SparkSession.builder().enableHiveSupport().getOrCreate()
sparkSession.sql("CREATE TABLE hive_features \
(feature_id BIGINT, \
feature_name STRING, \
feature_class STRING, \
state_alpha STRING, \
prim_lat_dec DOUBLE, \
prim_long_dec DOUBLE, \
elev_in_ft BIGINT) \
ROW FORMAT DELIMITED \
FIELDS TERMINATED BY '|' \
LINES TERMINATED BY '\n' \
LOCATION 's3://amzn-s3-demo-bucket/features';")
```
Sie haben jetzt eine aufgefüllte Hive-Tabelle mit Daten aus der `features.txt` Datei. Um zu überprüfen, ob sich Ihre Daten in der Tabelle befinden, führen Sie eine Spark-SQL-Abfrage aus, wie im folgenden Beispiel gezeigt.
```
sparkSession.sql(
"SELECT state_alpha, COUNT(*) FROM hive_features GROUP BY state_alpha;")
```
------
#### [ Hive ]
Führen Sie den folgenden Befehl aus, um eine Hive-Tabelle mit Hive zu erstellen.
```
CREATE TABLE hive_features
(feature_id BIGINT,
feature_name STRING ,
feature_class STRING ,
state_alpha STRING,
prim_lat_dec DOUBLE ,
prim_long_dec DOUBLE ,
elev_in_ft BIGINT)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
LINES TERMINATED BY '\n'
LOCATION 's3://amzn-s3-demo-bucket/features';
```
Sie haben jetzt eine Hive-Tabelle, die Daten aus der Datei enthält. `features.txt` Um zu überprüfen, ob sich Ihre Daten in der Tabelle befinden, führen Sie eine HiveQL-Abfrage aus, wie im folgenden Beispiel gezeigt.
```
SELECT state_alpha, COUNT(*) FROM hive_features GROUP BY state_alpha;
```
------
## Schritt 3: Daten nach DynamoDB kopieren
Verwenden Sie Spark oder Hive, um Daten in eine neue DynamoDB-Tabelle zu kopieren.
------
#### [ Spark ]
Um Daten aus der Hive-Tabelle, die Sie im vorherigen Schritt erstellt haben, nach DynamoDB zu kopieren, folgen Sie den **Schritten 1 bis 3** unter [Daten nach DynamoDB kopieren](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/EMRforDynamoDB.Tutorial.CopyDataToDDB.html). Dadurch wird eine neue DynamoDB-Tabelle mit dem Namen erstellt. `Features` Anschließend können Sie Daten direkt aus der Textdatei lesen und in Ihre DynamoDB-Tabelle kopieren, wie das folgende Beispiel zeigt.
```
import com.amazonaws.services.dynamodbv2.model.AttributeValue
import org.apache.hadoop.dynamodb.DynamoDBItemWritable
import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._
object EmrServerlessDynamoDbTest {
def main(args: Array[String]): Unit = {
jobConf.set("dynamodb.input.tableName", "Features")
jobConf.set("dynamodb.output.tableName", "Features")
jobConf.set("dynamodb.region", "region")
jobConf.set("mapred.output.format.class", "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
jobConf.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")
val rdd = sc.textFile("s3://amzn-s3-demo-bucket/ddb-connector/")
.map(row => {
val line = row.split("\\|")
val item = new DynamoDBItemWritable()
val elevInFt = if (line.length > 6) {
new AttributeValue().withN(line(6))
} else {
new AttributeValue().withNULL(true)
}
item.setItem(Map(
"feature_id" -> new AttributeValue().withN(line(0)),
"feature_name" -> new AttributeValue(line(1)),
"feature_class" -> new AttributeValue(line(2)),
"state_alpha" -> new AttributeValue(line(3)),
"prim_lat_dec" -> new AttributeValue().withN(line(4)),
"prim_long_dec" -> new AttributeValue().withN(line(5)),
"elev_in_ft" -> elevInFt)
.asJava)
(new Text(""), item)
})
rdd.saveAsHadoopDataset(jobConf)
}
}
```
------
#### [ Hive ]
Um Daten aus der Hive-Tabelle, die Sie im vorherigen Schritt erstellt haben, nach DynamoDB zu kopieren, folgen Sie den Anweisungen unter [Daten nach DynamoDB kopieren](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/EMRforDynamoDB.Tutorial.CopyDataToDDB.html).
------
## Schritt 4: Daten von DynamoDB abfragen
Verwenden Sie Spark oder Hive, um Ihre DynamoDB-Tabelle abzufragen.
------
#### [ Spark ]
Verwenden Sie entweder Spark SQL oder die Spark-API, um Daten aus der DynamoDB-Tabelle abzufragen, die Sie im vorherigen Schritt erstellt haben. MapReduce
**Example — Fragen Sie Ihre DynamoDB-Tabelle mit Spark SQL ab**
Die folgende Spark-SQL-Abfrage gibt eine Liste aller Feature-Typen in alphabetischer Reihenfolge zurück.
```
val dataFrame = sparkSession.sql("SELECT DISTINCT feature_class \
FROM ddb_features \
ORDER BY feature_class;")
```
Die folgende Spark-SQL-Abfrage gibt eine Liste aller Seen zurück, die mit dem Buchstaben *M* beginnen.
```
val dataFrame = sparkSession.sql("SELECT feature_name, state_alpha \
FROM ddb_features \
WHERE feature_class = 'Lake' \
AND feature_name LIKE 'M%' \
ORDER BY feature_name;")
```
Die folgende Spark-SQL-Abfrage gibt eine Liste aller Bundesstaaten mit mindestens drei Features zurück, die höher als eine Meile sind.
```
val dataFrame = sparkSession.dql("SELECT state_alpha, feature_class, COUNT(*) \
FROM ddb_features \
WHERE elev_in_ft > 5280 \
GROUP by state_alpha, feature_class \
HAVING COUNT(*) >= 3 \
ORDER BY state_alpha, feature_class;")
```
**Example — Fragen Sie Ihre DynamoDB-Tabelle mit der Spark-API ab MapReduce**
Die folgende MapReduce Abfrage gibt eine Liste aller Feature-Typen in alphabetischer Reihenfolge zurück.
```
val df = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
.map(pair => (pair._1, pair._2.getItem))
.map(pair => pair._2.get("feature_class").getS)
.distinct()
.sortBy(value => value)
.toDF("feature_class")
```
Die folgende MapReduce Abfrage gibt eine Liste aller Seen zurück, die mit dem Buchstaben *M* beginnen.
```
val df = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
.map(pair => (pair._1, pair._2.getItem))
.filter(pair => "Lake".equals(pair._2.get("feature_class").getS))
.filter(pair => pair._2.get("feature_name").getS.startsWith("M"))
.map(pair => (pair._2.get("feature_name").getS, pair._2.get("state_alpha").getS))
.sortBy(_._1)
.toDF("feature_name", "state_alpha")
```
Die folgende MapReduce Abfrage gibt eine Liste aller Bundesstaaten mit mindestens drei Features zurück, die höher als eine Meile sind.
```
val df = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
.map(pair => pair._2.getItem)
.filter(pair => pair.get("elev_in_ft").getN != null)
.filter(pair => Integer.parseInt(pair.get("elev_in_ft").getN) > 5280)
.groupBy(pair => (pair.get("state_alpha").getS, pair.get("feature_class").getS))
.filter(pair => pair._2.size >= 3)
.map(pair => (pair._1._1, pair._1._2, pair._2.size))
.sortBy(pair => (pair._1, pair._2))
.toDF("state_alpha", "feature_class", "count")
```
------
#### [ Hive ]
Um Daten aus der DynamoDB-Tabelle abzufragen, die Sie im vorherigen Schritt erstellt haben, folgen Sie den Anweisungen unter [Daten in der DynamoDB-Tabelle abfragen](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/EMRforDynamoDB.Tutorial.QueryDataInDynamoDB.html).
------