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.
# Beispiele und Tutorials für Studio-Notebooks in Managed Service für Apache Flink
**Topics**
+ [Tutorial: Erstellen Sie ein Studio-Notizbuch in Managed Service für Apache Flink](example-notebook.md)
+ [Tutorial: Stellen Sie ein Studio-Notebook als Managed Service für Apache Flink-Anwendung mit dauerhaftem Zustand bereit](example-notebook-deploy.md)
+ [Sehen Sie sich Beispielabfragen zur Analyse von Daten in einem Studio-Notizbuch an](how-zeppelin-sql-examples.md)
# Tutorial: Erstellen Sie ein Studio-Notizbuch in Managed Service für Apache Flink
Das folgende Tutorial zeigt, wie Sie ein Studio-Notebook erstellen, das Daten aus einem Kinesis-Datenstream oder einem Amazon MSK-Cluster liest.
**Topics**
+ [Erfüllen der Voraussetzungen](#example-notebook-setup)
+ [Erstellen Sie eine Datenbank AWS Glue](#example-notebook-glue)
+ [Nächste Schritte: Erstellen Sie ein Studio-Notizbuch mit Kinesis Data Streams oder Amazon MSK](#examples-notebook-nextsteps)
+ [Erstellen Sie ein Studio-Notebook mit Kinesis Data Streams](example-notebook-streams.md)
+ [Erstellen Sie ein Studio-Notebook mit Amazon MSK](example-notebook-msk.md)
+ [Bereinigen Sie Ihre Anwendung und die abhängigen Ressourcen](example-notebook-cleanup.md)
## Erfüllen der Voraussetzungen
Stellen Sie sicher, dass Sie AWS CLI Version 2 oder höher haben. Informationen zur Installation der neuesten AWS CLI Version finden Sie unter [Installation, Aktualisierung und Deinstallation der AWS CLI Version 2.](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)
## Erstellen Sie eine Datenbank AWS Glue
Ihr Studio-Notebook verwendet eine [AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/what-is-glue.html)-Datenbank für Metadaten zu Ihrer Amazon MSK-Datenquelle.
**Erstellen Sie eine AWS Glue Datenbank**
1. Öffnen Sie die AWS Glue Konsole unter [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue/).
1. Wählen Sie **Add database** (Datenbank hinzufügen). Geben Sie im Fenster **Datenbank hinzufügen** **default** als **Namen der Datenbank** ein. Wählen Sie **Erstellen** aus.
## Nächste Schritte: Erstellen Sie ein Studio-Notizbuch mit Kinesis Data Streams oder Amazon MSK
Mit diesem Tutorial können Sie ein Studio-Notebook erstellen, das entweder Kinesis Data Streams oder Amazon MSK verwendet:
+ [Erstellen Sie ein Studio-Notebook mit Kinesis Data Streams](example-notebook-streams.md): Mit Kinesis Data Streams können Sie schnell eine Anwendung erstellen, die einen Kinesis Data Stream als Quelle verwendet. Sie müssen nur einen Kinesis Data Stream als abhängige Ressource erstellen.
+ [Erstellen Sie ein Studio-Notebook mit Amazon MSK](example-notebook-msk.md): Mit Amazon MSK erstellen Sie eine Anwendung, die einen Amazon MSK-Cluster als Quelle verwendet. Sie müssen eine Amazon VPC, eine Amazon EC2-Client-Instance und einen Amazon MSK-Cluster als abhängige Ressourcen erstellen.
# Erstellen Sie ein Studio-Notebook mit Kinesis Data Streams
In diesem Tutorial wird beschrieben, wie Sie ein Studio-Notebook erstellen, das einen Kinesis Data Stream als Quelle verwendet.
**Topics**
+ [Erfüllen der Voraussetzungen](#example-notebook-streams-setup)
+ [Erstellen Sie eine Tabelle AWS Glue](#example-notebook-streams-glue)
+ [Erstellen Sie ein Studio-Notebook mit Kinesis Data Streams](#example-notebook-streams-create)
+ [Senden von Daten an den Kinesis Data Stream](#example-notebook-streams-send)
+ [Testen Sie Ihr Studio-Notebook](#example-notebook-streams-test)
## Erfüllen der Voraussetzungen
Bevor Sie ein Studio-Notebook erstellen, erstellen Sie einen Kinesis Data Stream (`ExampleInputStream`). Ihre Anwendung verwendet diesen Stream als Anwendungsquelle.
Sie können diesen Stream mithilfe der Amazon Kinesis-Konsole oder des folgenden AWS CLI -Befehls erstellen. Anweisungen für die Konsole finden Sie unter [Erstellen und Aktualisieren von Datenströmen](https://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html) im *Amazon Kinesis Data Streams Entwicklerhandbuch*. Benennen Sie den Stream **ExampleInputStream** und legen Sie die **Anzahl der offenen Shards** auf **1** fest.
Verwenden Sie den folgenden Amazon Kinesis `create-stream` AWS CLI Kinesis-Befehl AWS CLI, um den Stream (`ExampleInputStream`) mit dem zu erstellen.
```
$ aws kinesis create-stream \
--stream-name ExampleInputStream \
--shard-count 1 \
--region us-east-1 \
--profile adminuser
```
## Erstellen Sie eine Tabelle AWS Glue
Ihr Studio-Notebook verwendet eine [AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/what-is-glue.html)-Datenbank für Metadaten zu Ihrer Kinesis Data Streams-Datenquelle.
**Anmerkung**
Sie können die Datenbank entweder zuerst manuell erstellen oder sie beim Erstellen des Notebooks von Managed Service für Apache Flink für Sie erstellen lassen. Ebenso können Sie die Tabelle entweder manuell erstellen, wie im folgenden Abschnitt beschrieben, oder Sie können den Konnektorcode zum Erstellen einer Tabelle für Managed Service für Apache Flink in Ihrem Notebook innerhalb von Apache Zeppelin verwenden, um Ihre Tabelle über eine DDL-Anweisung zu erstellen. Sie können dann einchecken AWS Glue , um sicherzustellen, dass die Tabelle korrekt erstellt wurde.
**Erstellen einer Tabelle**
1. Melden Sie sich bei der an AWS-Managementkonsole und öffnen Sie die AWS Glue Konsole unter [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue/).
1. Wenn Sie noch keine AWS Glue Datenbank haben, wählen Sie in der linken Navigationsleiste **Datenbanken** aus. Wählen Sie **Datenbank hinzufügen**. Geben Sie im Fenster **Datenbank hinzufügen** **default** als **Namen der Datenbank** ein. Wählen Sie **Create (Erstellen)** aus.
1. Wählen Sie in der linken Navigationsleiste die Option **Tabellen**. Wählen Sie auf der Seite **Tabellen** die Optionen **Tabellen hinzufügen**, **Tabelle manuell hinzufügen** aus.
1. Geben Sie auf der Seite **Eigenschaften Ihrer Tabelle einrichten** **stock** als **Tabellennamen** ein. Stellen Sie sicher, dass Sie die Datenbank auswählen, die Sie zuvor erstellt haben. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Datenstore hinzufügen** die Option **Kinesis** aus. Geben Sie als **Streamnamen** **ExampleInputStream** ein. Wählen Sie für **Kinesis-Quell-URL** die Eingabetaste **https://kinesis.us-east-1.amazonaws.com**. Wenn Sie die **Kinesis-Quell-URL** kopieren und einfügen, achten Sie darauf, alle führenden oder nachfolgenden Leerzeichen zu löschen. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Klassifikation** die Option **JSON** aus. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Schema definieren** die Option „Spalte hinzufügen“, um eine Spalte hinzuzufügen. Fügen Sie Spalten mit den folgenden Eigenschaften hinzu:
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/managed-flink/latest/java/example-notebook-streams.html)
Wählen Sie **Weiter**.
1. Überprüfen Sie auf der nächsten Seite Ihre Einstellungen und wählen Sie **Fertigstellen.**
1. Wählen Sie die neu erstellte Tabelle aus der Liste der Tabellen aus.
1. Wählen Sie **Tabelle bearbeiten** und fügen Sie eine Eigenschaft mit dem Schlüssel `managed-flink.proctime` und dem Wert `proctime` hinzu.
1. Wählen Sie **Anwenden** aus.
## Erstellen Sie ein Studio-Notebook mit Kinesis Data Streams
Nachdem Sie die Ressourcen erstellt haben, die Ihre Anwendung verwendet, erstellen Sie Ihr Studio-Notebook.
**Topics**
+ [Erstellen Sie ein Studio-Notizbuch mit dem AWS-Managementkonsole](#example-notebook-create-streams-console)
+ [Erstellen Sie ein Studio-Notizbuch mit dem AWS CLI](#example-notebook-msk-create-api)
### Erstellen Sie ein Studio-Notizbuch mit dem AWS-Managementkonsole
1. Die Managed Service for Apache Flink-Konsole zu [ https://console.aws.amazon.com/managed-flink/Hause öffnen? region=us-east-1\$1/applications/dashboard](https://console.aws.amazon.com/managed-flink/home?region=us-east-1#/applications/dashboard).
1. Wählen Sie auf der Seite **Managed Service für Apache Flink-Anwendungen** die Registerkarte **Studio** aus. Wählen Sie **Studio-Notebook erstellen**.
**Anmerkung**
Sie können ein Studio-Notebook auch über die Amazon MSK- oder Kinesis Data Streams-Konsolen erstellen, indem Sie Ihren Amazon MSK-Eingabe-Cluster oder Kinesis Data Stream auswählen und dann **Daten in Echtzeit verarbeiten** auswählen.
1. Geben Sie auf der Seite **Notebook-Instance erstellen** folgende Informationen ein:
+ Geben Sie **MyNotebook** als Namen des Notebooks ein.
+ Wählen Sie **Standard** für die **AWS -Glue-Datenbank**.
Wählen Sie **Studio-Notebook erstellen**.
1. **Wählen Sie auf der Seite „Ausführen“ aus. **MyNotebook**** Warten Sie, bis der **Status** **Wird ausgeführt** angezeigt wird. Es fallen Gebühren an, wenn das Notebook läuft.
### Erstellen Sie ein Studio-Notizbuch mit dem AWS CLI
Gehen Sie wie folgt vor AWS CLI, um Ihr Studio-Notizbuch mit dem zu erstellen:
1. Überprüfen Sie die Konto-ID. Sie benötigen diesen Wert, um die Anwendung zu erstellen.
1. Erstellen Sie die Rolle `arn:aws:iam::AccountID:role/ZeppelinRole` und fügen Sie der automatisch erstellten Rolle über die Konsole die folgenden Berechtigungen hinzu.
`"kinesis:GetShardIterator",`
`"kinesis:GetRecords",`
`"kinesis:ListShards"`
1. Erstellen Sie eine Datei mit dem Namen `create.json` und den folgenden Inhalten. Ersetzen Sie die Platzhalterwerte durch Ihre Informationen.
```
{
"ApplicationName": "MyNotebook",
"RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
"ApplicationMode": "INTERACTIVE",
"ServiceExecutionRole": "arn:aws:iam::AccountID:role/ZeppelinRole",
"ApplicationConfiguration": {
"ApplicationSnapshotConfiguration": {
"SnapshotsEnabled": false
},
"ZeppelinApplicationConfiguration": {
"CatalogConfiguration": {
"GlueDataCatalogConfiguration": {
"DatabaseARN": "arn:aws:glue:us-east-1:AccountID:database/default"
}
}
}
}
}
```
1. Um Ihre Anwendung zu erstellen, führen Sie den folgenden Befehl aus.
```
aws kinesisanalyticsv2 create-application --cli-input-json file://create.json
```
1. Wenn der Befehl abgeschlossen ist, sehen Sie eine Ausgabe, die die Details für Ihr neues Studio-Notebook enthält. Es folgt ein Beispiel für die Ausgabe.
```
{
"ApplicationDetail": {
"ApplicationARN": "arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook",
"ApplicationName": "MyNotebook",
"RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
"ApplicationMode": "INTERACTIVE",
"ServiceExecutionRole": "arn:aws:iam::012345678901:role/ZeppelinRole",
...
```
1. Um Ihre Anwendung zu starten, führen Sie den folgenden Befehl aus. Ersetzen Sie die Beispielwerte durch Ihre Konto-ID.
```
aws kinesisanalyticsv2 start-application --application-arn arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook\
```
## Senden von Daten an den Kinesis Data Stream
Gehen Sie wie folgt vor, um Testdaten an Ihren Kinesis Data Stream zu senden:
1. Öffnen Sie den [Kinesis Data Generator](https://awslabs.github.io/amazon-kinesis-data-generator/web/help.html).
1. Wählen Sie „**Cognito-Benutzer erstellen mit CloudFormation**“.
1. Die CloudFormation Konsole wird mit der Kinesis Data Generator-Vorlage geöffnet. Wählen Sie **Weiter**.
1. Auf der Seite **Festlegen von Komponentendetails** geben Sie den Benutzernamen und das Passwort für Ihren Cognito-Benutzer ein. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Stack-Optionen konfigurieren** **Weiter** aus.
1. Wählen Sie auf der Seite „** Kinesis-Data-Generator-CognitoBenutzer überprüfen**“ die Option **Ich bestätige, dass AWS CloudFormation möglicherweise IAM-Ressourcen erstellt** werden. Kontrollkästchen. Wählen Sie **Stapel erstellen** aus.
1. Warten Sie, bis der CloudFormation Stapel fertig erstellt ist. **Nachdem der Stack abgeschlossen ist, öffnen Sie den **Kinesis-Data-Generator-Cognito-User-Stack** in der Konsole und wählen Sie die Registerkarte Outputs aus. CloudFormation ** **KinesisDataGeneratorUrl**Öffnen Sie die URL, die für den Ausgabewert aufgeführt ist.
1. Melden Sie sich auf der **Amazon Kinesis Data Generator**-Seite mit den Anmeldeinformationen an, die Sie in Schritt 4 erstellt haben.
1. Geben Sie auf der nächsten Seite die folgenden Werte an:
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/managed-flink/latest/java/example-notebook-streams.html)
Fügen Sie für **Datensatzvorlage** den folgenden Code ein:
```
{
"ticker": "{{random.arrayElement(
["AMZN","MSFT","GOOG"]
)}}",
"price": {{random.number(
{
"min":10,
"max":150
}
)}}
}
```
1. Wählen Sie **Daten senden** aus.
1. Der Generator sendet Daten an den Kinesis Data Stream.
Lassen Sie den Generator laufen, während Sie den nächsten Abschnitt abschließen.
## Testen Sie Ihr Studio-Notebook
In diesem Abschnitt verwenden Sie Ihr Studio-Notebook, um Daten aus Ihrem Kinesis Data Stream abzufragen.
1. Die Managed Service for Apache Flink-Konsole zu [ https://console.aws.amazon.com/managed-flink/Hause öffnen? region=us-east-1\$1/applications/dashboard](https://console.aws.amazon.com/managed-flink/home?region=us-east-1#/applications/dashboard).
1. Wählen Sie auf der Seite **Managed Service für Apache Flink-Anwendungen** die Registerkarte **Studio-Notebook** aus. Wählen Sie **MyNotebook**.
1. **Wählen Sie auf der Seite „In Apache Zeppelin öffnen“. **MyNotebook****
Die Oberfläche von Apache Zeppelin wird in einer neuen Registerkarte geöffnet.
1. Auf der Seite **Willkommen bei Zeppelin\$1** wählen Sie **Zeppelin Notiz** aus.
1. Geben Sie auf der Seite **Zeppelin Notiz** die folgende Abfrage in eine neue Notiz ein:
```
%flink.ssql(type=update)
select * from stock
```
Wählen Sie das Ausführungssymbol.
Nach kurzer Zeit werden in der Notiz Daten aus dem Kinesis Data Stream angezeigt.
Um das Apache Flink-Dashboard für Ihre Anwendung zu öffnen und betriebliche Aspekte zu sehen, wählen Sie **FLINK JOB**. Weitere Informationen zum Flink-Dashboard finden Sie unter [Apache Flink-Dashboard](https://docs.aws.amazon.com/managed-flink/latest/java/how-dashboard.html) im [Managed Service für Apache Flink Entwicklerhandbuch](https://docs.aws.amazon.com/).
Weitere Beispiele für Flink-Streaming-SQL-Abfragen finden Sie unter [Abfragen](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/sql/queries.html) in der [Apache Flink-Dokumentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/).
# Erstellen Sie ein Studio-Notebook mit Amazon MSK
In diesem Tutorial wird beschrieben, wie Sie ein Studio-Notebook erstellen, das einen Amazon-MSK-Cluster als Quelle verwendet.
**Topics**
+ [Einen Amazon MSK-Cluster einrichten](#example-notebook-msk-setup)
+ [Fügen Sie Ihrer VPC ein NAT-Gateway hinzu](#example-notebook-msk-nat)
+ [Erstellen Sie eine AWS Glue Verbindung und eine Tabelle](#example-notebook-msk-glue)
+ [Erstellen Sie ein Studio-Notebook mit Amazon MSK](#example-notebook-msk-create)
+ [Senden Sie Daten an Ihren Amazon MSK-Cluster](#example-notebook-msk-send)
+ [Testen Sie Ihr Studio-Notebook](#example-notebook-msk-test)
## Einen Amazon MSK-Cluster einrichten
Für dieses Tutorial benötigen Sie einen Amazon-MSK-Cluster, der Klartextzugriff ermöglicht. Wenn Sie noch keinen Amazon MSK-Cluster eingerichtet haben, folgen Sie dem Tutorial [Erste Schritte mit Amazon MSK, um eine Amazon](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) VPC, einen Amazon MSK-Cluster, ein Thema und eine Amazon-Client-Instance zu erstellen. EC2
Gehen Sie beim Befolgen des Tutorials wie folgt vor:
+ Ändern Sie in [Schritt 3: Amazon MSK-Cluster erstellen](https://docs.aws.amazon.com/msk/latest/developerguide/create-cluster.html) bei Schritt 4 den `ClientBroker`-Wert von `TLS` auf **PLAINTEXT**.
## Fügen Sie Ihrer VPC ein NAT-Gateway hinzu
Wenn Sie einen Amazon MSK-Cluster erstellt haben, indem Sie dem Tutorial [Erste Schritte mit Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) gefolgt sind, oder wenn Ihre bestehende Amazon VPC noch kein NAT-Gateway für ihre privaten Subnetze hat, müssen Sie Ihrer Amazon VPC ein NAT-Gateway hinzufügen. Das folgende Diagramm zeigt die Architektur.
![\[AWS VPC architecture with public and private subnets, NAT gateway, and Glue Data Catalog integration.\]](http://docs.aws.amazon.com/de_de/managed-flink/latest/java/images/vpc_05.png)
Gehen Sie wie folgt vor, um ein NAT-Gateway für Ihre Amazon VPC zu erstellen:
1. Öffnen Sie die Amazon-VPC-Konsole unter [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Wählen Sie in der linken Navigationsleiste **NAT-Gateway** aus.
1. Wählen Sie auf der Seite **NAT-Gateways** die Option **NAT-Gateway erstellen** aus.
1. Geben Sie auf der Seite **NAT-Gateway erstellen** die folgenden Werte an:
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/managed-flink/latest/java/example-notebook-msk.html)
Wählen Sie **NAT-Gateway erstellen** aus.
1. Wählen Sie in der linken Navigationsleiste **Routing-Tabellen** aus.
1. Klicken Sie auf **Create Route Table (Routing-Tabelle erstellen)**.
1. Geben Sie auf der Seite **Routing-Tabelle erstellen** folgende Informationen ein:
+ **Name-Tag:** **ZeppelinRouteTable**
+ **VPC**: Wählen Sie Ihre VPC (z. B. **AWS KafkaTutorialVPC**).
Wählen Sie **Create (Erstellen)** aus.
1. Wählen Sie in der Liste der Routentabellen. **ZeppelinRouteTable** Klicken Sie auf der Registerkarte **Routen** auf **Routen bearbeiten**.
1. Wählen Sie auf der Seite **Routen bearbeiten** die Option **Route hinzufügen** aus.
1. Geben Sie im ****Für-**Ziel** **0.0.0.0/0** ein. Wählen Sie für **Target** die Option **NAT Gateway**, **ZeppelinGateway**. Wählen Sie **Routen speichern** aus. Klicken Sie auf **Close** (Schließen).
1. Wählen Sie auf der Seite Routing-Tabellen die **ZeppelinRouteTable**Option „**Subnetzzuordnungen**“ aus. Wählen Sie **Subnetzzuordnungen bearbeiten** aus.
1. **Wählen Sie auf der Seite **Subnetzzuordnungen bearbeiten** die Optionen **AWS KafkaTutorialSubnet2 und AWS KafkaTutorialSubnet 3** aus.** Wählen Sie **Save (Speichern)** aus.
## Erstellen Sie eine AWS Glue Verbindung und eine Tabelle
Ihr Studio-Notebook verwendet eine [AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/what-is-glue.html)-Datenbank für Metadaten zu Ihrer Amazon MSK-Datenquelle. In diesem Abschnitt erstellen Sie eine AWS Glue Verbindung, die beschreibt, wie Sie auf Ihren Amazon MSK-Cluster zugreifen können, und eine AWS Glue Tabelle, die beschreibt, wie Sie die Daten in Ihrer Datenquelle für Clients wie Ihr Studio-Notebook präsentieren.
**Eine Verbindung erstellen**
1. Melden Sie sich bei der an AWS-Managementkonsole und öffnen Sie die AWS Glue Konsole unter [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue/).
1. Wenn Sie noch keine AWS Glue Datenbank haben, wählen Sie in der linken Navigationsleiste **Datenbanken** aus. Wählen Sie **Datenbank hinzufügen**. Geben Sie im Fenster **Datenbank hinzufügen** **default** als **Namen der Datenbank** ein. Wählen Sie **Create (Erstellen)** aus.
1. Wählen Sie in der linken Navigationsleiste **Verbindungen** aus. Wählen Sie **Verbindung hinzufügen** aus.
1. Geben Sie im Fenster **Verbindung hinzufügen** die folgenden Werte ein:
+ Geben Sie für **Verbindungsname** **ZeppelinConnection** ein.
+ Wählen Sie für **Verbindungstyp** den Eintrag **Kafka**.
+ Geben Sie für den **Kafka-Bootstrap-Server URLs** die Bootstrap-Broker-String für Ihren Cluster an. Sie können die Bootstrap-Broker entweder über die MSK-Konsole oder durch Eingabe des folgenden CLI-Befehls abrufen:
```
aws kafka get-bootstrap-brokers --region us-east-1 --cluster-arn ClusterArn
```
+ Deaktivieren Sie das Kontrollkästchen **SSL-Verbindung erforderlich**.
Wählen Sie **Weiter**.
1. Geben Sie auf der **VPC**-Seite die folgenden Werte an:
+ Wählen Sie für **VPC** den Namen Ihrer VPC (z. B. ** AWS KafkaTutorialVPC**).
+ **Wählen Sie für **Subnetz 2** aus.AWS KafkaTutorialSubnet**
+ Wählen Sie für **Sicherheitsgruppen** alle verfügbaren Gruppen aus.
Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Verbindungseigenschaften** / **Verbindungszugriff** die Option **Fertigstellen** aus.
**Erstellen einer Tabelle**
**Anmerkung**
Sie können die Tabelle entweder manuell erstellen, wie in den folgenden Schritten beschrieben, oder Sie können den Konnektorcode zum Erstellen einer Tabelle für Managed Service für Apache Flink in Ihrem Notebook innerhalb von Apache Zeppelin verwenden, um Ihre Tabelle über eine DDL-Anweisung zu erstellen. Sie können dann einchecken AWS Glue , um sicherzustellen, dass die Tabelle korrekt erstellt wurde.
1. Wählen Sie in der linken Navigationsleiste die Option **Tabellen**. Wählen Sie auf der Seite **Tabellen** die Optionen **Tabellen hinzufügen**, **Tabelle manuell hinzufügen** aus.
1. Geben Sie auf der Seite **Eigenschaften Ihrer Tabelle einrichten** **stock** als **Tabellennamen** ein. Stellen Sie sicher, dass Sie die Datenbank auswählen, die Sie zuvor erstellt haben. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Datenspeicher hinzufügen** die Option **Kafka** aus. Geben Sie als **Themennamen** Ihren Themennamen ein (z. B. **AWS KafkaTutorialTopic**). Wählen Sie für **Verbindung **ZeppelinConnection****.
1. Wählen Sie auf der Seite **Klassifikation** die Option **JSON** aus. Wählen Sie **Weiter**.
1. Wählen Sie auf der Seite **Schema definieren** die Option „Spalte hinzufügen“, um eine Spalte hinzuzufügen. Fügen Sie Spalten mit den folgenden Eigenschaften hinzu:
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/de_de/managed-flink/latest/java/example-notebook-msk.html)
Wählen Sie **Weiter**.
1. Überprüfen Sie auf der nächsten Seite Ihre Einstellungen und wählen Sie **Fertigstellen.**
1. Wählen Sie die neu erstellte Tabelle aus der Liste der Tabellen aus.
1. Wählen Sie **Tabelle bearbeiten** und fügen Sie die folgenden Eigenschaften hinzu:
+ Schlüssel:`managed-flink.proctime`, Wert: `proctime`
+ Schlüssel:`flink.properties.group.id`, Wert: `test-consumer-group`
+ Schlüssel:`flink.properties.auto.offset.reset`, Wert: `latest`
+ Schlüssel:`classification`, Wert: `json`
Ohne diese Schlüssel/Wert-Paare tritt im Flink-Notebook ein Fehler auf.
1. Wählen Sie **Anwenden** aus.
## Erstellen Sie ein Studio-Notebook mit Amazon MSK
Nachdem Sie die Ressourcen erstellt haben, die Ihre Anwendung verwendet, erstellen Sie Ihr Studio-Notebook.
**Topics**
+ [Erstellen Sie ein Studio-Notizbuch mit dem AWS-Managementkonsole](#example-notebook-create-msk-console)
+ [Erstellen Sie ein Studio-Notizbuch mit dem AWS CLI](#example-notebook-msk-create-api)
**Anmerkung**
Sie können ein Studio-Notebook auch von der Amazon MSK-Konsole aus erstellen, indem Sie einen vorhandenen Cluster auswählen und dann **Daten in Echtzeit verarbeiten** wählen.
### Erstellen Sie ein Studio-Notizbuch mit dem AWS-Managementkonsole
1. Die Managed Service for Apache Flink-Konsole zu [ https://console.aws.amazon.com/managed-flink/Hause öffnen? region=us-east-1\$1/applications/dashboard](https://console.aws.amazon.com/managed-flink/home?region=us-east-1#/applications/dashboard).
1. Wählen Sie auf der Seite **Managed Service für Apache Flink-Anwendungen** die Registerkarte **Studio** aus. Wählen Sie **Studio-Notebook erstellen**.
**Anmerkung**
Um ein Studio-Notebook über die Amazon MSK- oder Kinesis Data Streams-Konsolen zu erstellen, wählen Sie Ihren Amazon MSK-Eingabe-Cluster oder Kinesis Data Stream aus und wählen Sie dann **Daten in Echtzeit verarbeiten** aus.
1. Geben Sie auf der Seite **Notebook-Instance erstellen** folgende Informationen ein:
+ Geben Sie **MyNotebook** als **Studio-Notebookname**.
+ Wählen Sie **Standard** für die **AWS -Glue-Datenbank**.
Wählen Sie **Studio-Notebook erstellen**.
1. **Wählen Sie auf der Seite die Registerkarte Konfiguration aus. **MyNotebook**** Wählen Sie im Abschnitt **Netzwerk** die Option **Bearbeiten**.
1. Wählen Sie auf der MyNotebook Seite **Netzwerk bearbeiten für** die **VPC-Konfiguration basierend auf dem Amazon MSK-Cluster** aus. Wählen Sie Ihren Amazon MSK-Cluster für den **Amazon MSK-Cluster** aus. Wählen Sie **Änderungen speichern**.
1. **Wählen Sie auf der **MyNotebook**Seite die Option Ausführen aus.** Warten Sie, bis der **Status** **Wird ausgeführt** angezeigt wird.
### Erstellen Sie ein Studio-Notizbuch mit dem AWS CLI
Gehen Sie wie folgt vor AWS CLI, um Ihr Studio-Notizbuch mit dem zu erstellen:
1. Stellen Sie sicher, dass Sie über die folgenden Informationen verfügen: Sie benötigen diese Werte, um Ihre Anwendung zu erstellen.
+ Ihre Konto-ID.
+ Das Subnetz IDs und die Sicherheitsgruppen-ID für die Amazon VPC, die Ihren Amazon MSK-Cluster enthält.
1. Erstellen Sie eine Datei mit dem Namen `create.json` und den folgenden Inhalten. Ersetzen Sie die Platzhalterwerte durch Ihre Informationen.
```
{
"ApplicationName": "MyNotebook",
"RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
"ApplicationMode": "INTERACTIVE",
"ServiceExecutionRole": "arn:aws:iam::AccountID:role/ZeppelinRole",
"ApplicationConfiguration": {
"ApplicationSnapshotConfiguration": {
"SnapshotsEnabled": false
},
"VpcConfigurations": [
{
"SubnetIds": [
"SubnetID 1",
"SubnetID 2",
"SubnetID 3"
],
"SecurityGroupIds": [
"VPC Security Group ID"
]
}
],
"ZeppelinApplicationConfiguration": {
"CatalogConfiguration": {
"GlueDataCatalogConfiguration": {
"DatabaseARN": "arn:aws:glue:us-east-1:AccountID:database/default"
}
}
}
}
}
```
1. Um Ihre Anwendung zu erstellen, führen Sie den folgenden Befehl aus.
```
aws kinesisanalyticsv2 create-application --cli-input-json file://create.json
```
1. Wenn der Befehl abgeschlossen ist, sollte eine Ausgabe wie die folgende angezeigt werden, die die Details für Ihr neues Studio-Notebook enthält:
```
{
"ApplicationDetail": {
"ApplicationARN": "arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook",
"ApplicationName": "MyNotebook",
"RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
"ApplicationMode": "INTERACTIVE",
"ServiceExecutionRole": "arn:aws:iam::012345678901:role/ZeppelinRole",
...
```
1. Um Ihre Anwendung zu starten, führen Sie den folgenden Befehl aus. Ersetzen Sie die Beispielwerte durch Ihre Konto-ID.
```
aws kinesisanalyticsv2 start-application --application-arn arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook\
```
## Senden Sie Daten an Ihren Amazon MSK-Cluster
In diesem Abschnitt führen Sie ein Python-Skript in Ihrem EC2 Amazon-Client aus, um Daten an Ihre Amazon MSK-Datenquelle zu senden.
1. Connect zu Ihrem EC2 Amazon-Client her.
1. Führen Sie die folgenden Befehle aus, um Python Version 3, Pip und das Kafka für Python-Paket zu installieren, und bestätigen Sie die Aktionen:
```
sudo yum install python37
curl -O https://bootstrap.pypa.io/get-pip.py
python3 get-pip.py --user
pip install kafka-python
```
1. Konfigurieren Sie das AWS CLI auf Ihrem Client-Computer, indem Sie den folgenden Befehl eingeben:
```
aws configure
```
Geben Sie Ihre Kontoanmeldeinformationen ein, und **us-east-1** für die `region`.
1. Erstellen Sie eine Datei mit dem Namen `stock.py` und den folgenden Inhalten. Ersetzen Sie den Beispielwert durch die Bootstrap Brokers-Zeichenfolge Ihres Amazon MSK-Clusters und aktualisieren Sie den Themennamen, falls Ihr Thema nicht: **AWS KafkaTutorialTopic**
```
from kafka import KafkaProducer
import json
import random
from datetime import datetime
BROKERS = "<>"
producer = KafkaProducer(
bootstrap_servers=BROKERS,
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
retry_backoff_ms=500,
request_timeout_ms=20000,
security_protocol='PLAINTEXT')
def getStock():
data = {}
now = datetime.now()
str_now = now.strftime("%Y-%m-%d %H:%M:%S")
data['event_time'] = str_now
data['ticker'] = random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV'])
price = random.random() * 100
data['price'] = round(price, 2)
return data
while True:
data =getStock()
# print(data)
try:
future = producer.send("AWSKafkaTutorialTopic", value=data)
producer.flush()
record_metadata = future.get(timeout=10)
print("sent event to Kafka! topic {} partition {} offset {}".format(record_metadata.topic, record_metadata.partition, record_metadata.offset))
except Exception as e:
print(e.with_traceback())
```
1. Führen Sie das Skript mit dem folgenden Befehl aus:
```
$ python3 stock.py
```
1. Lassen Sie das Skript laufen, während Sie den folgenden Abschnitt abschließen.
## Testen Sie Ihr Studio-Notebook
In diesem Abschnitt verwenden Sie Ihr Studio-Notebook, um Daten aus Ihrem Amazon MSK-Cluster abzufragen.
1. [Die Managed Service for Apache Flink-Konsole zu Hause öffnen? https://console.aws.amazon.com/managed-flink/ region=us-east-1\$1/applications/dashboard](https://console.aws.amazon.com/managed-flink/home?region=us-east-1#/applications/dashboard).
1. Wählen Sie auf der Seite **Managed Service für Apache Flink-Anwendungen** die Registerkarte **Studio-Notebook** aus. Wählen Sie **MyNotebook**.
1. **Wählen Sie auf der Seite „In Apache Zeppelin öffnen“. **MyNotebook****
Die Oberfläche von Apache Zeppelin wird in einer neuen Registerkarte geöffnet.
1. Auf der Seite **Willkommen bei Zeppelin\$1** wählen Sie **Zeppelin neue Notiz** aus.
1. Geben Sie auf der Seite **Zeppelin Notiz** die folgende Abfrage in eine neue Notiz ein:
```
%flink.ssql(type=update)
select * from stock
```
Wählen Sie das Ausführungssymbol.
Die Anwendung zeigt Daten aus dem Amazon MSK-Cluster an.
Um das Apache Flink-Dashboard für Ihre Anwendung zu öffnen und betriebliche Aspekte zu sehen, wählen Sie **FLINK JOB**. Weitere Informationen zum Flink-Dashboard finden Sie unter [Apache Flink-Dashboard](https://docs.aws.amazon.com/managed-flink/latest/java/how-dashboard.html) im [Managed Service für Apache Flink Entwicklerhandbuch](https://docs.aws.amazon.com/).
Weitere Beispiele für Flink-Streaming-SQL-Abfragen finden Sie unter [Abfragen](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/sql/queries.html) in der [Apache Flink-Dokumentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/).
# Bereinigen Sie Ihre Anwendung und die abhängigen Ressourcen
## Löschen Sie Ihr Studio-Notebook
1. Öffnen Sie die Managed Service für Apache Flink-Konsole.
1. Wählen Sie **MyNotebook**.
1. Wählen Sie **Aktionen** und dann **Löschen** aus.
## Löschen Sie Ihre AWS Glue Datenbank und Ihre Verbindung
1. Öffnen Sie die AWS Glue Konsole unter [https://console.aws.amazon.com/glue/](https://console.aws.amazon.com/glue/).
1. Wählen Sie in der linken Navigationsleiste **Datenbanken** aus. Markieren Sie das Kontrollkästchen neben **Standard**, um es auszuwählen. Wählen Sie **Aktion**, **Datenbank löschen**. Bestätigen Sie Ihre Auswahl.
1. Wählen Sie in der linken Navigationsleiste **Verbindungen** aus. Markieren Sie das Kontrollkästchen neben, **ZeppelinConnection**um es auszuwählen. Wählen Sie **Aktion**, **Verbindung löschen**. Bestätigen Sie Ihre Auswahl.
## So löschen Sie die IAM-Rolle und -Richtlinie
1. Öffnen Sie unter [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/) die IAM-Konsole.
1. Klicken Sie in der linken Navigationsleiste auf **Rollen**.
1. Verwenden Sie die Suchleiste, um nach der **ZeppelinRole**Rolle zu suchen.
1. Wählen Sie die **ZeppelinRole**Rolle aus. Wählen Sie **Rolle löschen** aus. Bestätigen Sie das Löschen.
## Löschen Sie Ihre CloudWatch Protokollgruppe
Die Konsole erstellt eine CloudWatch Logs-Gruppe und einen Log-Stream für Sie, wenn Sie Ihre Anwendung mit der Konsole erstellen. Sie haben keine Protokollgruppe und keinen Stream, wenn Sie Ihre Anwendung mit der AWS CLI erstellt haben.
1. Öffnen Sie die CloudWatch Konsole unter [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/).
1. Wählen Sie in der linken Navigationsleiste **Protokollgruppen** aus.
1. Wählen Sie die Gruppe**/AWS/KinesisAnalytics/MyNotebook**log aus.
1. Wählen Sie **Actions (Aktionen)**, **Delete log group(s) (Protokollgruppe(n) löschen)** aus. Bestätigen Sie das Löschen.
## Kinesis Data Streams Streams-Ressourcen bereinigen
Um Ihren Kinesis Stream zu löschen, öffnen Sie die Kinesis Data Streams-Konsole, wählen Sie Ihren Kinesis Stream aus und wählen Sie **Aktionen**, **Löschen**.
## Bereinigen von MSK-Ressourcen
Führen Sie die Schritte in diesem Abschnitt aus, wenn Sie für dieses Tutorial einen Amazon MSK-Cluster erstellt haben. Dieser Abschnitt enthält Anweisungen zur Bereinigung Ihrer Amazon EC2-Client-Instance, Amazon VPC und Amazon MSK-Cluster.
### Löschen Sie Ihren Amazon MSK-Cluster
Gehen Sie wie folgt vor, wenn Sie für dieses Tutorial einen Amazon MSK-Cluster erstellt haben.
1. Die Amazon MSK-Konsole zu [https://console.aws.amazon.com/msk/Hause öffnen? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/).
1. Wählen Sie **AWS KafkaTutorialCluster**. Wählen Sie **Löschen** aus. Geben Sie **delete** in das angezeigte Fenster ein und bestätigen Sie Ihre Auswahl.
### Beenden Ihrer Client-Instance
Gehen Sie wie folgt vor, wenn Sie für dieses Tutorial eine Amazon EC2-Client-Instance erstellt haben.
1. Öffnen Sie die Amazon-EC2-Konsole unter [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).
1. Wählen Sie in der linken Navigationsleiste **Instances** aus.
1. Klicken Sie auf das Kästchen neben, um es auszuwählen. **ZeppelinClient**
1. Wählen Sie **Instance-Status**, **Instance beenden**.
### Löschen Ihrer Amazon VPC
Gehen Sie wie folgt vor, wenn Sie für dieses Tutorial eine Amazon VPC erstellt haben.
1. Öffnen Sie die Amazon-EC2-Konsole unter [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).
1. Wählen Sie in der linken Navigationsleiste **Netzwerkschnittstellen** aus.
1. Geben Sie Ihre VPC-ID in das Suchfeld ein und drücken Sie die Eingabetaste.
1. Aktivieren Sie das Kontrollkästchen in der Kopfzeile der Tabelle, um alle angezeigten Netzwerkschnittstellen auszuwählen.
1. Wählen Sie **Actions (Aktionen)**, **Loslösen (Detach)**. Wählen Sie in dem daraufhin angezeigten Fenster unter **Trennung erzwingen** die Option **Aktivieren** aus. Wählen Sie **Trennen** und warten Sie, bis alle Netzwerkschnittstellen den Status **Verfügbar** erreichen.
1. Aktivieren Sie das Kontrollkästchen in der Kopfzeile der Tabelle, um alle angezeigten Netzwerkschnittstellen erneut auszuwählen.
1. Wählen Sie **Aktionen**, **Löschen** aus. Bestätigen Sie die Aktion.
1. Öffnen Sie die Amazon-VPC-Konsole unter [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
1. Wählen Sie **AWS KafkaTutorialVPC** aus. Wählen Sie **Aktionen**, **VPC löschen** aus. Geben Sie **delete** ein und bestätigen Sie den Löschvorgang.
# Tutorial: Stellen Sie ein Studio-Notebook als Managed Service für Apache Flink-Anwendung mit dauerhaftem Zustand bereit
Das folgende Tutorial zeigt, wie Sie ein Studio-Notebook als Managed-Service für Apache Flink-Anwendung mit einem dauerhaften Status bereitstellen.
**Topics**
+ [Erfüllen der Voraussetzungen](#example-notebook-durable-setup)
+ [Stellen Sie eine Anwendung mit einem dauerhaften Status bereit, indem Sie den AWS-Managementkonsole](#example-notebook-deploy-console)
+ [Stellen Sie eine Anwendung mit einem dauerhaften Zustand bereit, indem Sie AWS CLI](#example-notebook-deploy-cli)
## Erfüllen der Voraussetzungen
Erstellen Sie ein neues Studio-Notizbuch, indem Sie den Anweisungen folgen[Tutorial: Erstellen Sie ein Studio-Notizbuch in Managed Service für Apache Flink](example-notebook.md) und entweder Kinesis Datenstrom oder Amazon MSK verwenden. Name des Studio-Notizbuchs `ExampleTestDeploy`.
## Stellen Sie eine Anwendung mit einem dauerhaften Status bereit, indem Sie den AWS-Managementkonsole
1. Fügen Sie einen S3-Bucket-Speicherort hinzu, an dem der gepackte Code unter **Speicherort des Anwendungscodes gespeichert werden soll - *optional*** in der Konsole. Dies ermöglicht die Schritte zum Bereitstellen und Ausführen Ihrer Anwendung direkt vom Notebook aus.
1. Fügen Sie der Anwendungsrolle die erforderlichen Berechtigungen hinzu, um die von Ihnen verwendete Rolle zum Lesen und Schreiben in einen Amazon S3-Bucket zu aktivieren und um eine Managed Service for Apache Flink-Anwendung zu starten:
+ Amazon S3 FullAccess
+ Amazon hat es geschafft- flinkFullAccess
+ Zugriff auf Ihre Quellen, Ziele und VPCs gegebenenfalls Weitere Informationen finden Sie unter [Überprüfen Sie die IAM-Berechtigungen für Studio-Notebooks](how-zeppelin-iam.md).
1. Verwenden Sie den folgenden Beispielcode:
```
%flink.ssql(type=update)
CREATE TABLE exampleoutput (
'ticket' VARCHAR,
'price' DOUBLE
)
WITH (
'connector' = 'kinesis',
'stream' = 'ExampleOutputStream',
'aws.region' = 'us-east-1',
'scan.stream.initpos' = 'LATEST',
'format' = 'json'
);
INSERT INTO exampleoutput SELECT ticker, price FROM exampleinputstream
```
1. Mit der Einführung dieses Feature sehen Sie in der rechten oberen Ecke jeder Notiz in Ihrem Notizbuch ein neues Dropdown-Menü mit dem Namen des Notizbuchs. Sie haben die folgenden Möglichkeiten:
+ Sehen Sie sich die Studio-Notizbucheinstellungen in der AWS-Managementkonsole an.
+ Erstellen Sie Ihren Zeppelin Note und exportieren Sie ihn zu Amazon S3. Geben Sie an dieser Stelle einen Namen für Ihre Anwendung ein und wählen Sie **Erstellen und Exportieren**. Sie erhalten eine Benachrichtigung, wenn der Export abgeschlossen ist.
+ Bei Bedarf können Sie alle zusätzlichen Tests der ausführbaren Datei in Amazon S3 anzeigen und ausführen.
+ Sobald der Build abgeschlossen ist, können Sie Ihren Code als Kinesis-Streaming-Anwendung mit dauerhaftem Zustand und automatischer Skalierung bereitstellen.
+ Verwenden Sie das Drop-down-Menü und wählen Sie **Zeppelin Note als Kinesis-Streaming-Anwendung bereitstellen**. Überprüfen Sie den Namen der Anwendung und wählen Sie **Über AWS Konsole bereitstellen** aus.
+ Dadurch gelangen Sie zu der AWS-Managementkonsole Seite zum Erstellen einer Managed Service for Apache Flink-Anwendung. Beachten Sie, dass Anwendungsname, Parallelität, Codespeicherort, Standard-Glue-DB, VPC (falls zutreffend) und IAM-Rollen vorab ausgefüllt wurden. Stellen Sie sicher, dass die IAM-Rollen über die erforderlichen Berechtigungen für Ihre Quellen und Ziele verfügen. Snapshots sind standardmäßig aktiviert, um eine dauerhafte Verwaltung des Anwendungsstatus zu gewährleisten.
+ Wählen Sie **Erstellen der Anwendung**.
+ Sie können alle Einstellungen **konfigurieren** und ändern und anschließend **Ausführen** wählen, um Ihre Streaming-Anwendung zu starten.
## Stellen Sie eine Anwendung mit einem dauerhaften Zustand bereit, indem Sie AWS CLI
Um eine Anwendung mithilfe von bereitzustellen AWS CLI, müssen Sie Ihr System so aktualisieren, AWS CLI dass es das mit Ihren Beta 2-Informationen bereitgestellte Servicemodell verwendet. Weitere Informationen zur Verwendung des aktualisierten Servicemodells finden Sie unter [Erfüllen der VoraussetzungenErfüllen der Voraussetzungen](example-notebook.md#example-notebook-setup).
Der folgende Beispielcode erstellt ein neues Studio-Notizbuch:
```
aws kinesisanalyticsv2 create-application \
--application-name \
--runtime-environment ZEPPELIN-FLINK-3_0 \
--application-mode INTERACTIVE \
--service-execution-role
--application-configuration '{
"ZeppelinApplicationConfiguration": {
"CatalogConfiguration": {
"GlueDataCatalogConfiguration": {
"DatabaseARN": "arn:aws:glue:us-east-1::database/"
}
}
},
"FlinkApplicationConfiguration": {
"ParallelismConfiguration": {
"ConfigurationType": "CUSTOM",
"Parallelism": 4,
"ParallelismPerKPU": 4
}
},
"DeployAsApplicationConfiguration": {
"S3ContentLocation": {
"BucketARN": "arn:aws:s3:::",
"BasePath": "/something/"
}
},
"VpcConfigurations": [
{
"SecurityGroupIds": [
""
],
"SubnetIds": [
"",
""
]
}
]
}' \
--region us-east-1
```
Das folgende Codebeispiel startet ein Studio-Notebook:
```
aws kinesisanalyticsv2 start-application \
--application-name \
--region us-east-1 \
--no-verify-ssl
```
Der folgende Code gibt die URL für die Apache Zeppelin-Notizbuch-Seite einer Anwendung zurück:
```
aws kinesisanalyticsv2 create-application-presigned-url \
--application-name \
--url-type ZEPPELIN_UI_URL \
--region us-east-1 \
--no-verify-ssl
```
# Sehen Sie sich Beispielabfragen zur Analyse von Daten in einem Studio-Notizbuch an
**Topics**
+ [Erstellen Sie Tabellen mit Amazon MSK/Apache Kafka](#how-zeppelin-examples-creating-tables)
+ [Erstellen Sie Tabellen mit Kinesis](#how-zeppelin-examples-creating-tables-with-kinesis)
+ [Fragen Sie ein taumelndes Fenster ab](#how-zeppelin-examples-tumbling)
+ [Fragen Sie ein Schiebefenster ab](#how-zeppelin-examples-sliding)
+ [Verwenden Sie interaktives SQL](#how-zeppelin-examples-interactive-sql)
+ [Verwenden Sie den BlackHole SQL-Konnektor](#how-zeppelin-examples-blackhole-connector-sql)
+ [Verwenden Sie Scala, um Beispieldaten zu generieren](#notebook-example-data-generator)
+ [Verwenden Sie interaktives Scala](#notebook-example-interactive-scala)
+ [Interaktives Python verwenden](#notebook-example-interactive-python)
+ [Verwenden Sie eine Kombination aus interaktivem Python, SQL und Scala](#notebook-example-interactive-pythonsqlscala)
+ [Verwenden Sie einen kontoübergreifenden Kinesis-Datenstream](#notebook-example-crossaccount-kds)
Informationen zu den SQL-Abfrageeinstellungen von Apache Flink finden Sie unter [ Flink auf Zeppelin-Notebooks für interaktive Datenanalyse](https://flink.apache.org/ecosystem/2020/06/23/flink-on-zeppelin-part2.html).
Um Ihre Anwendung im Apache-Flink-Dashboard anzuzeigen, wählen Sie **FLINK-AUFTRAG** auf der Seite **Zeppelin Notiz** Ihrer Anwendung.
Weitere Informationen zu Fensterabfragen finden Sie unter [Windows](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/stream/operators/windows.html) in der [Apache-Flink-Dokumentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/).
Weitere Beispiele für Streaming-SQL-Abfragen in Apache Flink finden Sie unter [Abfragen](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/sql/queries.html) in der [Apache-Flink-Dokumentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/).
## Erstellen Sie Tabellen mit Amazon MSK/Apache Kafka
Sie können den Amazon-MSK-Flink-Konnektor mit Managed Service für Apache Flink Studio verwenden, um Ihre Verbindung mit Klartext-, SSL- oder IAM-Authentifizierung zu authentifizieren. Erstellen Sie Ihre Tabellen mit den spezifischen Eigenschaften gemäß Ihren Anforderungen.
```
-- Plaintext connection
CREATE TABLE your_table (
`column1` STRING,
`column2` BIGINT
) WITH (
'connector' = 'kafka',
'topic' = 'your_topic',
'properties.bootstrap.servers' = '',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
);
-- SSL connection
CREATE TABLE your_table (
`column1` STRING,
`column2` BIGINT
) WITH (
'connector' = 'kafka',
'topic' = 'your_topic',
'properties.bootstrap.servers' = '',
'properties.security.protocol' = 'SSL',
'properties.ssl.truststore.location' = '/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts',
'properties.ssl.truststore.password' = 'changeit',
'properties.group.id' = 'myGroup',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
);
-- IAM connection (or for MSK Serverless)
CREATE TABLE your_table (
`column1` STRING,
`column2` BIGINT
) WITH (
'connector' = 'kafka',
'topic' = 'your_topic',
'properties.bootstrap.servers' = '',
'properties.security.protocol' = 'SASL_SSL',
'properties.sasl.mechanism' = 'AWS_MSK_IAM',
'properties.sasl.jaas.config' = 'software.amazon.msk.auth.iam.IAMLoginModule required;',
'properties.sasl.client.callback.handler.class' = 'software.amazon.msk.auth.iam.IAMClientCallbackHandler',
'properties.group.id' = 'myGroup',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
);
```
Sie können diese mit anderen Eigenschaften im [Apache-Kafka-SQL-Konnektor](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/kafka/) kombinieren.
## Erstellen Sie Tabellen mit Kinesis
Im folgenden Beispiel erstellen Sie eine Tabelle mit Kinesis:
```
CREATE TABLE KinesisTable (
`column1` BIGINT,
`column2` BIGINT,
`column3` BIGINT,
`column4` STRING,
`ts` TIMESTAMP(3)
)
PARTITIONED BY (column1, column2)
WITH (
'connector' = 'kinesis',
'stream' = 'test_stream',
'aws.region' = '',
'scan.stream.initpos' = 'LATEST',
'format' = 'csv'
);
```
Weitere Informationen zu anderen Eigenschaften, die Sie verwenden können, finden Sie unter [Amazon Kinesis Data Streams SQL-Konnektor](https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/kinesis/).
## Fragen Sie ein taumelndes Fenster ab
Die folgende Flink-Streaming-SQL-Abfrage wählt den höchsten Preis in jedem fünfsekündigen rollierenden Fenster aus der `ZeppelinTopic`-Tabelle aus:
```
%flink.ssql(type=update)
SELECT TUMBLE_END(event_time, INTERVAL '5' SECOND) as winend, MAX(price) as five_second_high, ticker
FROM ZeppelinTopic
GROUP BY ticker, TUMBLE(event_time, INTERVAL '5' SECOND)
```
## Fragen Sie ein Schiebefenster ab
Die folgende Flink-Streaming-SQL-Abfrage wählt den höchsten Preis in jedem fünfsekündigen rollierenden Fenster aus der `ZeppelinTopic`-Tabelle aus:
```
%flink.ssql(type=update)
SELECT HOP_END(event_time, INTERVAL '3' SECOND, INTERVAL '5' SECOND) AS winend, MAX(price) AS sliding_five_second_max
FROM ZeppelinTopic//or your table name in AWS Glue
GROUP BY HOP(event_time, INTERVAL '3' SECOND, INTERVAL '5' SECOND)
```
## Verwenden Sie interaktives SQL
In diesem Beispiel wird der Höchstwert der Ereignis- und Verarbeitungszeit sowie die Summe der Werte aus der Schlüssel-Wert-Tabelle ausgegeben. Stellen Sie sicher, dass Sie das Beispielskript zur Datengenerierung aus dem laufenden [Verwenden Sie Scala, um Beispieldaten zu generieren](#notebook-example-data-generator) haben. Informationen zum Ausprobieren anderer SQL-Abfragen wie Filtern und Joins in Ihrem Studio-Notebook finden Sie in der Apache-Flink-Dokumentation: [Abfragen](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/sql/queries.html) in der Apache-Flink-Dokumentation.
```
%flink.ssql(type=single, parallelism=4, refreshInterval=1000, template={2}
records seen until Processing Time: {1}
and Event Time: {0}
)
-- An interactive query prints how many records from the `key-value-stream` we have seen so far, along with the current processing and event time.
SELECT
MAX(`et`) as `et`,
MAX(`pt`) as `pt`,
SUM(`value`) as `sum`
FROM
`key-values`
```
```
%flink.ssql(type=update, parallelism=4, refreshInterval=1000)
-- An interactive tumbling window query that displays the number of records observed per (event time) second.
-- Browse through the chart views to see different visualizations of the streaming result.
SELECT
TUMBLE_START(`et`, INTERVAL '1' SECONDS) as `window`,
`key`,
SUM(`value`) as `sum`
FROM
`key-values`
GROUP BY
TUMBLE(`et`, INTERVAL '1' SECONDS),
`key`;
```
## Verwenden Sie den BlackHole SQL-Konnektor
Der BlackHole SQL-Connector erfordert nicht, dass Sie einen Kinesis-Datenstream oder einen Amazon MSK-Cluster erstellen, um Ihre Abfragen zu testen. Informationen zum SQL-Konnektor finden Sie unter BlackHole [BlackHole SQL Connector](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/connectors/blackhole.html) in der Apache Flink-Dokumentation. In diesem Beispiel ist der Standardkatalog ein speicherinterner Katalog.
```
%flink.ssql
CREATE TABLE default_catalog.default_database.blackhole_table (
`key` BIGINT,
`value` BIGINT,
`et` TIMESTAMP(3)
) WITH (
'connector' = 'blackhole'
)
```
```
%flink.ssql(parallelism=1)
INSERT INTO `test-target`
SELECT
`key`,
`value`,
`et`
FROM
`test-source`
WHERE
`key` > 3
```
```
%flink.ssql(parallelism=2)
INSERT INTO `default_catalog`.`default_database`.`blackhole_table`
SELECT
`key`,
`value`,
`et`
FROM
`test-target`
WHERE
`key` > 7
```
## Verwenden Sie Scala, um Beispieldaten zu generieren
In diesem Beispiel wird Scala verwendet, um Beispieldaten zu generieren. Sie können diese Beispieldaten verwenden, um verschiedene Abfragen zu testen. Verwenden Sie die Anweisung create table, um die Schlüssel-Wert-Tabelle zu erstellen.
```
import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource
import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator
import org.apache.flink.streaming.api.scala.DataStream
import java.sql.Timestamp
// ad-hoc convenience methods to be defined on Table
implicit class TableOps[T](table: DataStream[T]) {
def asView(name: String): DataStream[T] = {
if (stenv.listTemporaryViews.contains(name)) {
stenv.dropTemporaryView("`" + name + "`")
}
stenv.createTemporaryView("`" + name + "`", table)
return table;
}
}
```
```
%flink(parallelism=4)
val stream = senv
.addSource(new DataGeneratorSource(RandomGenerator.intGenerator(1, 10), 1000))
.map(key => (key, 1, new Timestamp(System.currentTimeMillis)))
.asView("key-values-data-generator")
```
```
%flink.ssql(parallelism=4)
-- no need to define the paragraph type with explicit parallelism (such as "%flink.ssql(parallelism=2)")
-- in this case the INSERT query will inherit the parallelism of the of the above paragraph
INSERT INTO `key-values`
SELECT
`_1` as `key`,
`_2` as `value`,
`_3` as `et`
FROM
`key-values-data-generator`
```
## Verwenden Sie interaktives Scala
Dies ist die Scala-Übersetzung von [Verwenden Sie interaktives SQL](#how-zeppelin-examples-interactive-sql). Weitere Scala-Beispiele finden Sie unter [Tabellen-API](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/tableApi.html) in der Apache-Flink-Dokumentation.
```
%flink
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
// ad-hoc convenience methods to be defined on Table
implicit class TableOps(table: Table) {
def asView(name: String): Table = {
if (stenv.listTemporaryViews.contains(name)) {
stenv.dropTemporaryView(name)
}
stenv.createTemporaryView(name, table)
return table;
}
}
```
```
%flink(parallelism=4)
// A view that computes many records from the `key-values` we have seen so far, along with the current processing and event time.
val query01 = stenv
.from("`key-values`")
.select(
$"et".max().as("et"),
$"pt".max().as("pt"),
$"value".sum().as("sum")
).asView("query01")
```
```
%flink.ssql(type=single, parallelism=16, refreshInterval=1000, template={2}
records seen until Processing Time: {1}
and Event Time: {0}
)
-- An interactive query prints the query01 output.
SELECT * FROM query01
```
```
%flink(parallelism=4)
// An tumbling window view that displays the number of records observed per (event time) second.
val query02 = stenv
.from("`key-values`")
.window(Tumble over 1.seconds on $"et" as $"w")
.groupBy($"w", $"key")
.select(
$"w".start.as("window"),
$"key",
$"value".sum().as("sum")
).asView("query02")
```
```
%flink.ssql(type=update, parallelism=4, refreshInterval=1000)
-- An interactive query prints the query02 output.
-- Browse through the chart views to see different visualizations of the streaming result.
SELECT * FROM `query02`
```
## Interaktives Python verwenden
Dies ist die Python-Übersetzung von [Verwenden Sie interaktives SQL](#how-zeppelin-examples-interactive-sql). Weitere Python-Beispiele finden Sie unter [Tabellen-API](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/tableApi.html) in der Apache-Flink-Dokumentation.
```
%flink.pyflink
from pyflink.table.table import Table
def as_view(table, name):
if (name in st_env.list_temporary_views()):
st_env.drop_temporary_view(name)
st_env.create_temporary_view(name, table)
return table
Table.as_view = as_view
```
```
%flink.pyflink(parallelism=16)
# A view that computes many records from the `key-values` we have seen so far, along with the current processing and event time
st_env \
.from_path("`keyvalues`") \
.select(", ".join([
"max(et) as et",
"max(pt) as pt",
"sum(value) as sum"
])) \
.as_view("query01")
```
```
%flink.ssql(type=single, parallelism=16, refreshInterval=1000, template={2}
records seen until Processing Time: {1}
and Event Time: {0}
)
-- An interactive query prints the query01 output.
SELECT * FROM query01
```
```
%flink.pyflink(parallelism=16)
# A view that computes many records from the `key-values` we have seen so far, along with the current processing and event time
st_env \
.from_path("`key-values`") \
.window(Tumble.over("1.seconds").on("et").alias("w")) \
.group_by("w, key") \
.select(", ".join([
"w.start as window",
"key",
"sum(value) as sum"
])) \
.as_view("query02")
```
```
%flink.ssql(type=update, parallelism=16, refreshInterval=1000)
-- An interactive query prints the query02 output.
-- Browse through the chart views to see different visualizations of the streaming result.
SELECT * FROM `query02`
```
## Verwenden Sie eine Kombination aus interaktivem Python, SQL und Scala
Sie können eine beliebige Kombination aus SQL, Python und Scala in Ihrem Notebook für interaktive Analysen verwenden. In einem Studio-Notebook, das Sie als dauerhafte Anwendung bereitstellen möchten, können Sie eine Kombination aus SQL und Scala verwenden. Dieses Beispiel zeigt Ihnen die Abschnitte, die ignoriert werden, und diejenigen, die in der Anwendung mit dem dauerhaften Zustand bereitgestellt werden.
```
%flink.ssql
CREATE TABLE `default_catalog`.`default_database`.`my-test-source` (
`key` BIGINT NOT NULL,
`value` BIGINT NOT NULL,
`et` TIMESTAMP(3) NOT NULL,
`pt` AS PROCTIME(),
WATERMARK FOR `et` AS `et` - INTERVAL '5' SECOND
)
WITH (
'connector' = 'kinesis',
'stream' = 'kda-notebook-example-test-source-stream',
'aws.region' = 'eu-west-1',
'scan.stream.initpos' = 'LATEST',
'format' = 'json',
'json.timestamp-format.standard' = 'ISO-8601'
)
```
```
%flink.ssql
CREATE TABLE `default_catalog`.`default_database`.`my-test-target` (
`key` BIGINT NOT NULL,
`value` BIGINT NOT NULL,
`et` TIMESTAMP(3) NOT NULL,
`pt` AS PROCTIME(),
WATERMARK FOR `et` AS `et` - INTERVAL '5' SECOND
)
WITH (
'connector' = 'kinesis',
'stream' = 'kda-notebook-example-test-target-stream',
'aws.region' = 'eu-west-1',
'scan.stream.initpos' = 'LATEST',
'format' = 'json',
'json.timestamp-format.standard' = 'ISO-8601'
)
```
```
%flink()
// ad-hoc convenience methods to be defined on Table
implicit class TableOps(table: Table) {
def asView(name: String): Table = {
if (stenv.listTemporaryViews.contains(name)) {
stenv.dropTemporaryView(name)
}
stenv.createTemporaryView(name, table)
return table;
}
}
```
```
%flink(parallelism=1)
val table = stenv
.from("`default_catalog`.`default_database`.`my-test-source`")
.select($"key", $"value", $"et")
.filter($"key" > 10)
.asView("query01")
```
```
%flink.ssql(parallelism=1)
-- forward data
INSERT INTO `default_catalog`.`default_database`.`my-test-target`
SELECT * FROM `query01`
```
```
%flink.ssql(type=update, parallelism=1, refreshInterval=1000)
-- forward data to local stream (ignored when deployed as application)
SELECT * FROM `query01`
```
```
%flink
// tell me the meaning of life (ignored when deployed as application!)
print("42!")
```
## Verwenden Sie einen kontoübergreifenden Kinesis-Datenstream
Um einen Kinesis-Datenstrom zu verwenden, der sich in einem anderen Konto als dem Konto befindet, das Ihr Studio-Notebook enthält, erstellen Sie eine Serviceausführungsrolle in dem Konto, in dem Ihr Studio-Notebook ausgeführt wird, und eine Rollenvertrauensrichtlinie für das Konto, das den Datenstrom enthält. Verwenden Sie `aws.credentials.provider`, `aws.credentials.role.arn` und `aws.credentials.role.sessionName` im Kinesis-Konnektor in Ihrer DDL-Anweisung create table, um eine Tabelle anhand des Datenstroms zu erstellen.
Verwenden Sie die folgende Serviceausführungsrolle für das Studio-Notebook-Konto.
```
{
"Sid": "AllowNotebookToAssumeRole",
"Effect": "Allow",
"Action": "sts:AssumeRole"
"Resource": "*"
}
```
Verwenden Sie die `AmazonKinesisFullAccess`-Richtlinie und die folgende Rollenvertrauensrichtlinie für das Datenstrom-Konto.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::123456789012:root"
},
"Action": "sts:AssumeRole",
"Condition": {}
}
]
}
```
------
Verwenden Sie den folgenden Absatz für die create-table-Anweisung.
```
%flink.ssql
CREATE TABLE test1 (
name VARCHAR,
age BIGINT
) WITH (
'connector' = 'kinesis',
'stream' = 'stream-assume-role-test',
'aws.region' = 'us-east-1',
'aws.credentials.provider' = 'ASSUME_ROLE',
'aws.credentials.role.arn' = 'arn:aws:iam:::role/stream-assume-role-test-role',
'aws.credentials.role.sessionName' = 'stream-assume-role-test-session',
'scan.stream.initpos' = 'TRIM_HORIZON',
'format' = 'json'
)
```