Schritt 2. Implementieren der Transformationslogik

Anmerkung

Benutzerdefinierte visuelle Transformationen unterstützen nur Python-Skripte. Scala wird nicht unterstützt.

Um den Code hinzuzufügen, der die durch die JSON-Konfigurationsdatei definierte Funktion implementiert, wird empfohlen, die Python-Datei am selben Speicherort wie die JSON-Datei und unter demselben Namen abzulegen, aber mit der Erweiterung „PY“. AWS Glue Studio koppelt die JSON- und PY-Dateien automatisch, sodass Sie den Pfad der Python-Datei nicht in der Konfigurationsdatei angeben müssen.

Fügen Sie in der Python-Datei die deklarierte Funktion mit den konfigurierten benannten Parametern hinzu und registrieren Sie sie zur Verwendung in DynamicFrame. Das Folgende ist ein Beispiel für eine Python-Datei:

from awsglue import DynamicFrame

# self refers to the DynamicFrame to transform, 
# the parameter names must match the ones defined in the config
# if it's optional, need to provide a default value
def myTransform(self, email, phone, age=None, gender="", 
                      country="", promotion=False):
   resulting_dynf = # do some transformation on self 
   return resulting_dynf
   
DynamicFrame.myTransform = myTransform
        

Es wird empfohlen, ein AWS Glue-Notebook zu verwenden, um den Python-Code am schnellsten zu entwickeln und zu testen. Siehe Erste Schritte mit Notebooks in AWS Glue Studio.

Um zu veranschaulichen, wie die Transformationslogik implementiert wird, ist die benutzerdefinierte visuelle Transformation im folgenden Beispiel eine Transformation zum Filtern eingehender Daten. Dabei werden nur die Daten berücksichtigt, die sich auf einen bestimmten US-Bundesstaat beziehen. Die JSON-Datei enthält den Parameter für functionName als custom_filter_state und zwei Argumente („state“ und „colName“ mit Typ „str“).

Die JSON-Beispielkonfigurationsdatei lautet:

{
"name": "custom_filter_state",
"displayName": "Filter State",
"description": "A simple example to filter the data to keep only the state indicated.",
"functionName": "custom_filter_state",
"parameters": [
   {
    "name": "colName",
    "displayName": "Column name",
    "type": "str",
    "description": "Name of the column in the data that holds the state postal code"
   },
   {
    "name": "state",
    "displayName": "State postal code",
    "type": "str",
    "description": "The postal code of the state whole rows to keep"
   }   
  ]
}
        

So implementieren Sie das Begleitskript in Python

1. Starten Sie ein AWS Glue-Notebook und führen Sie die Anfangszelle aus, die für die zu startende Sitzung bereitgestellt wird. Durch Ausführen der Anfangszelle werden die erforderlichen Basiskomponenten erstellt.

2. Erstellen Sie eine Funktion, die die im Beispiel beschriebene Filterung durchführt, und registrieren Sie sie auf DynamicFrame. Kopieren Sie den unten stehenden Code und fügen Sie ihn in eine Zelle im AWS Glue-Notebook ein.

   from awsglue import DynamicFrame
   
   def custom_filter_state(self, colName, state):
       return self.filter(lambda row: row[colName] == state)
   
   DynamicFrame.custom_filter_state = custom_filter_state                    
                   

3. Erstellen oder laden Sie Beispieldaten, um den Code in derselben Zelle oder in einer neuen Zelle zu testen. Wenn Sie die Beispieldaten in einer neuen Zelle hinzufügen, vergessen Sie nicht, die Zelle auszuführen. Beispielsweise:

   # A few of rows of sample data to test
   data_sample = [
       {"state": "CA", "count": 4},
       {"state": "NY", "count": 2},
       {"state": "WA", "count": 3}    
   ]
   df1 = glueContext.sparkSession.sparkContext.parallelize(data_sample).toDF()
   dynf1 = DynamicFrame.fromDF(df1, glueContext, None)                    
                   

4. Führen Sie einen Test durch, um den „custom_filter_state“ mit verschiedenen Argumenten zu validieren:

   

5. Nachdem Sie mehrere Tests ausgeführt haben, speichern Sie den Code mit der PY-Erweiterung und benennen Sie die PY-Datei mit einem Namen, der den Namen der JSON-Datei widerspiegelt. Die PY- und JSON-Dateien sollten sich im selben Transformationsordner befinden.

   Kopieren Sie den folgenden Code, fügen Sie ihn in eine Datei ein und benennen Sie ihn mit einer PY-Dateierweiterung um.

   from awsglue import DynamicFrame
   
   def custom_filter_state(self, colName, state):
       return self.filter(lambda row: row[colName] == state)
   
   DynamicFrame.custom_filter_state = custom_filter_state                    
                   

6. Öffnen Sie in AWS Glue Studio einen visuellen Auftrag und fügen Sie die Transformation dem Auftrag hinzu, indem Sie sie aus der Liste der verfügbaren Transforms (Transformationen) auswählen.

   Um diese Transformation in einem Python-Skriptcode wiederzuverwenden, fügen Sie den Amazon-S3-Pfad zur PY-Datei im Auftrag unter „Referenzierter Dateipfad“ hinzu und importieren Sie im Skript den Namen der Python-Datei (ohne die Erweiterung), indem Sie ihn am Anfang der Datei hinzufügen. Zum Beispiel: import <Name der Datei (ohne die Erweiterung)>