AWS Glue DynamicFrameClasse Scala

Pacchetto: com.amazonaws.services.glue

class DynamicFrame extends Serializable with Logging  (
           val glueContext : GlueContext,
           _records : RDD[DynamicRecord],
           val name : String = s"",
           val transformationContext : String = DynamicFrame.UNDEFINED,
           callSite : CallSite = CallSite("Not provided", ""),
           stageThreshold : Long = 0,
           totalThreshold : Long = 0,
           prevErrors : => Long = 0,
           errorExpr : => Unit = {} )

Un DynamicFrame è una raccolta distribuita di oggetti DynamicRecord autodescrittivi.

DynamicFrame sono stati progettati per fornire un modello di dati flessibile per le operazioni ETL (estrazione, trasformazione e caricamento). Questi oggetti non richiedono la creazione di uno schema e possono essere usati per leggere e trasformare i dati che contengono valori e tipi non organizzati e non coerenti. Un schema può essere calcolato on demand per le operazioni che ne richiedono uno.

DynamicFrame offrono un'ampia gamma di trasformazioni per la pulizia dei dati e operazioni ETL. Supportano anche la conversione da e verso SparkSQL DataFrames per l'integrazione con il codice esistente e le numerose operazioni di analisi che DataFrames forniscono.

I parametri seguenti vengono condivisi tra molte trasformazioni AWS Glue che costituiscono oggetti DynamicFrame:

Val errorsCount

val errorsCount

Numero di record di errore in questo oggetto DynamicFrame. Include gli errori restituiti dalle operazioni precedenti.

Def applyMapping

def applyMapping( mappings : Seq[Product4[String, String, String, String]],
                  caseSensitive : Boolean = true,
                  transformationContext : String = "",
                  callSite : CallSite = CallSite("Not provided", ""),
                  stageThreshold : Long = 0,
                  totalThreshold : Long = 0
                ) : DynamicFrame 

Seleziona, proietta e trasmette le colonne in base a una sequenza di mappature.

Ogni mappatura è costituita da una colonna e un tipo di origine e da una colonna e un tipo target. Le mappature possono essere specificate come un 4-tuple (source_path, source_type, target_path, target_type) o un oggetto MappingSpec contenente le stesse informazioni.

Oltre che per semplici proiezioni e trasferimenti, le mappature possono essere usate per annidare campi o annullarne l'annidamento separando i componenti del percorso con "'." (punto).

Ad esempio, supponiamo di avere un DynamicFrame con lo schema seguente.

 {{{
   root
   |-- name: string
   |-- age: int
   |-- address: struct
   |    |-- state: string
   |    |-- zip: int
 }}}

Puoi effettuare la chiamata seguente per annullare l'annidamento dei campi state e zip.

 {{{
   df.applyMapping(
    Seq(("name", "string", "name", "string"),
        ("age", "int", "age", "int"),
        ("address.state", "string", "state", "string"),
        ("address.zip", "int", "zip", "int")))
 }}}

Lo schema risultante è il seguente.

 {{{
   root
   |-- name: string
   |-- age: int
   |-- state: string
   |-- zip: int
 }}}

Puoi anche usare applyMapping per riannidare le colonne. Ad esempio, il codice seguente inverte la trasformazione precedente e crea una struttura denominata address nel target.

 {{{
   df.applyMapping(
    Seq(("name", "string", "name", "string"),
        ("age", "int", "age", "int"),
        ("state", "string", "address.state", "string"),
        ("zip", "int", "address.zip", "int")))
 }}}

I nomi dei campi che contengono i caratteri "." (periodo) possono essere quotati utilizzando le virgolette (``).

Def assertErrorThreshold

def assertErrorThreshold : Unit

Operazione che forza il calcolo e verifica che il numero di record di errore sia inferiore a stageThreshold e totalThreshold. Genera un'eccezione se una delle due condizioni non è vera.

Def conteggio

lazy
def count

Restituisce il numero di elementi inclusi in questo oggetto DynamicFrame.

Def dropField

def dropField( path : String,
               transformationContext : String = "",
               callSite : CallSite = CallSite("Not provided", ""),
               stageThreshold : Long = 0,
               totalThreshold : Long = 0
             ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame con la colonna specificata rimossa.

Def dropFields

def dropFields( fieldNames : Seq[String],   // The column names to drop.
                transformationContext : String = "",
                callSite : CallSite = CallSite("Not provided", ""),
                stageThreshold : Long = 0,
                totalThreshold : Long = 0
              ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame con le colonne specificate rimosse.

Questo metodo può essere usato per eliminare colonne annidate, incluse quelle all'interno di matrici, ma non per eliminare elementi di matrice specifici.

Def dropNulls

def dropNulls( transformationContext : String = "",
               callSite : CallSite = CallSite("Not provided", ""),
               stageThreshold : Long = 0,
               totalThreshold : Long = 0 )

Restituisce un nuovo oggetto DynamicFrame con tutte le colonne null rimosse.

Cornice Def errorsAsDynamic

def errorsAsDynamicFrame

Restituisce un nuovo oggetto DynamicFrame contenente i record degli errori da questo DynamicFrame.

Def filtro

def filter( f : DynamicRecord => Boolean,
            errorMsg : String = "",
            transformationContext : String = "",
            callSite : CallSite = CallSite("Not provided"),
            stageThreshold : Long = 0,
            totalThreshold : Long = 0
          ) : DynamicFrame 

Crea un nuovo oggetto DynamicFrame contenente solo i record per i quali la funzione "f" restituisce true. La funzione di filtro "f" non dovrebbe modificare il record di input.

Def getName

def getName : String 

Restituisce il nome di questo oggetto DynamicFrame.

Def getNumPartitions

def getNumPartitions

Restituisce il numero di partizioni incluse in questo oggetto DynamicFrame.

Def calcolato getSchemaIf

def getSchemaIfComputed : Option[Schema] 

Restituisce lo schema se è già stato calcolato. Non analizza i dati se lo schema non è stato ancora calcolato.

Def isSchemaComputed

def isSchemaComputed : Boolean 

Restituisce true se lo schema è stato calcolato per questo oggetto DynamicFrame oppure restituisce false in caso contrario. Se questo metodo restituisce false, la chiamata del metodo schema richiede un altro passaggio sui record in questo oggetto DynamicFrame.

Def javaToPython

def javaToPython : JavaRDD[Array[Byte]] 

Def join

def join( keys1 : Seq[String],
          keys2 : Seq[String],
          frame2 : DynamicFrame,
          transformationContext : String = "",
          callSite : CallSite = CallSite("Not provided", ""),
          stageThreshold : Long = 0,
          totalThreshold : Long = 0
        ) : DynamicFrame 

Restituisce il risultato dell'esecuzione di una query equijoin con frame2 usando le chiavi specificate.

Def mappa

def map( f : DynamicRecord => DynamicRecord,
         errorMsg : String = "",
         transformationContext : String = "",
         callSite : CallSite = CallSite("Not provided", ""),
         stageThreshold : Long = 0,
         totalThreshold : Long = 0
       ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame creato applicando la funzione "f" specificata a ogni record in questo oggetto DynamicFrame.

Questo metodo copia ogni record prima di applicare la funzione specificata e di conseguenza è sicuro per la modifica dei record. Se la funzione di mappatura genera un'eccezione per un determinato record, il record verrà contrassegnato come errore e l'analisi dello stack verrà salvata come colonna nel record di errore.

Def mergeDynamicFrames

def mergeDynamicFrames( stageDynamicFrame: DynamicFrame,  primaryKeys: Seq[String], transformationContext: String = "",
                         options: JsonOptions = JsonOptions.empty, callSite: CallSite = CallSite("Not provided"),
                         stageThreshold: Long = 0, totalThreshold: Long = 0): DynamicFrame 

Unisce questo DynamicFrame con un DynamicFrame temporaneo basato sulle chiavi primarie specificate per identificare i record. I registri duplicati (registri con le stesse chiavi primarie) non vengono deduplicati. Se non è presente alcun record corrispondente nel frame temporaneo, tutti i record (inclusi i duplicati) vengono mantenuti dall'origine. Se il frame di staging dispone di record corrispondenti, i suoi registri sovrascrivono i registri nell'origine in AWS Glue.

Il DynamicFrame restituito contiene il record A in questi casi:

Il frame di origine e il frame temporaneo non devono avere lo stesso schema.

val mergedFrame: DynamicFrame = srcFrame.mergeDynamicFrames(stageFrame, Seq("id1", "id2"))

Def printSchema

def printSchema : Unit 

Stampa lo schema di questo oggetto DynamicFrame in stdout in un formato leggibile.

Def recomputeSchema

def recomputeSchema : Schema 

Forza un nuovo calcolo dello schema. Questa operazione richiede una scansione dei dati, ma potrebbe "limitare" lo schema se lo schema corrente include dati che nono sono presenti nei dati.

Restituisce lo schema ricalcolato.

Def relazionalizzazione

def relationalize( rootTableName : String,
                   stagingPath : String,
                   options : JsonOptions = JsonOptions.empty,
                   transformationContext : String = "",
                   callSite : CallSite = CallSite("Not provided"),
                   stageThreshold : Long = 0,
                   totalThreshold : Long = 0
                 ) : Seq[DynamicFrame] 

Appiattisce tutte le strutture annidate e trasforma tramite pivoting le matrici in tabelle separate.

Questa operazione può essere usata per preparare dati annidati a più livelli per l'inserimento in un database relazionale. Le strutture annidate vengono appiattite allo stesso modo della trasformazione Unnest. Inoltre, le matrici vengono trasformate tramite pivoting in tabelle separate attraverso un'operazione in cui ogni elemento di matrice diventa una riga. Ad esempio, supponiamo di avere un DynamicFrame con i dati seguenti.

 {"name": "Nancy", "age": 47, "friends": ["Fred", "Lakshmi"]}
 {"name": "Stephanie", "age": 28, "friends": ["Yao", "Phil", "Alvin"]}
 {"name": "Nathan", "age": 54, "friends": ["Nicolai", "Karen"]}

Eseguire il seguente codice.

{{{
  df.relationalize("people", "s3:/my_bucket/my_path", JsonOptions.empty)
}}}

Il codice produce due tabelle. La prima tabella è denominata "persone" e contiene quanto segue.

{{{
  {"name": "Nancy", "age": 47, "friends": 1}
  {"name": "Stephanie", "age": 28, "friends": 2}
  {"name": "Nathan", "age": 54, "friends": 3)
}}}

Qui le matrici friends sono state sostituite con una chiave di join generata automaticamente. Viene creata una tabella separata denominata people.friends con il contenuto seguente.

{{{
  {"id": 1, "index": 0, "val": "Fred"}
  {"id": 1, "index": 1, "val": "Lakshmi"}
  {"id": 2, "index": 0, "val": "Yao"}
  {"id": 2, "index": 1, "val": "Phil"}
  {"id": 2, "index": 2, "val": "Alvin"}
  {"id": 3, "index": 0, "val": "Nicolai"}
  {"id": 3, "index": 1, "val": "Karen"}
}}}

In questa tabella "id" è una chiave di join che identifica il record da cui proviene l'elemento della matrice, "index" fa riferimento alla posizione nella matrice originale e "val" è l'effettiva voce della matrice.

Il metodo relationalize restituisce la sequenza di oggetti DynamicFrame creati applicando questo processo in modo ricorsivo a tutte le matrici.

Def renameField

def renameField( oldName : String,
                 newName : String,
                 transformationContext : String = "",
                 callSite : CallSite = CallSite("Not provided", ""),
                 stageThreshold : Long = 0,
                 totalThreshold : Long = 0
               ) : DynamicFrame 

Restituisce un nuovo DynamicFrame con il campo specificato rinominato.

Questo metodo può essere usato per rinominare campi annidati. Ad esempio, il codice seguente rinominerebbe state in state_code all'interno della struttura address.

{{{
  df.renameField("address.state", "address.state_code")
}}}

Def ripartizione

def repartition( numPartitions : Int,
                 transformationContext : String = "",
                 callSite : CallSite = CallSite("Not provided", ""),
                 stageThreshold : Long = 0,
                 totalThreshold : Long = 0
               ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame con partizioni numPartitions.

Def resolveChoice

def resolveChoice( specs : Seq[Product2[String, String]] = Seq.empty[ResolveSpec],
                   choiceOption : Option[ChoiceOption] = None,
                   database : Option[String] = None,
                   tableName : Option[String] = None,
                   transformationContext : String = "",
                   callSite : CallSite = CallSite("Not provided", ""),
                   stageThreshold : Long = 0,
                   totalThreshold : Long = 0
                 ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame sostituendo uno o più oggetti ChoiceType con un tipo più specifico.

Ci sono due modi per utilizzare resolveChoice. Il primo consiste nell'indicare una sequenza di colonne specifiche e come risolverle. Queste vengono specificate come tuple costituite da coppie (colonna, operazione).

Sono disponibili le operazioni seguenti:

L'altra modalità per resolveChoice è specificare una singola risoluzione per tutti gli oggetti ChoiceType. Puoi usare questa modalità nei casi in cui l'elenco completo di oggetti ChoiceType non è noto prima dell'esecuzione. Oltre alle operazioni elencate in precedenza, questa modalità supporta anche l'operazione seguente:

Esempi:

Risoluzione della colonna user.id mediante casting a un tipo int, facendo in modo che il campo address mantenga solo le strutture:

{{{
  df.resolveChoice(specs = Seq(("user.id", "cast:int"), ("address", "project:struct")))
}}}

Risoluzione di tutti gli oggetti ChoiceType mediante conversione di ogni scelta in una colonna separata:

{{{
  df.resolveChoice(choiceOption = Some(ChoiceOption("make_cols")))
}}}

Risoluzione di tutte gli oggetti ChoiceType mediante casting ai tipi nella tabella del catalogo specificata:

{{{
  df.resolveChoice(choiceOption = Some(ChoiceOption("match_catalog")),
                   database = Some("my_database"),
                   tableName = Some("my_table"))
}}}

Def schema

def schema : Schema 

Restituisce lo schema di questo oggetto DynamicFrame.

Lo schema restituito è garantito per contenere ogni campo presente in un record in questo DynamicFrame. Tuttavia, in un esiguo numero di casi, può anche contenere campi aggiuntivi. Puoi utilizzare il metodo Unnest per "ridurre" lo schema in base ai record in questo DynamicFrame.

Def selectField

def selectField( fieldName : String,
                 transformationContext : String = "",
                 callSite : CallSite = CallSite("Not provided", ""),
                 stageThreshold : Long = 0,
                 totalThreshold : Long = 0
               ) : DynamicFrame 

Restituisce un singolo campo come DynamicFrame.

Def selectFields

def selectFields( paths : Seq[String],
                  transformationContext : String = "",
                  callSite : CallSite = CallSite("Not provided", ""),
                  stageThreshold : Long = 0,
                  totalThreshold : Long = 0
                ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame contenente le colonne specificate.

Def mostra

def show( numRows : Int = 20 ) : Unit 

Stampa le righe di questo oggetto DynamicFrame in formato JSON.

Def semplifica DDBJSON

Le esportazioni DynamoDB con il connettore di esportazione AWS Glue DynamoDB producono file JSON di strutture annidate specifiche. Per ulteriori informazioni, consulta Data objects. simplifyDDBJson Semplifica le colonne annidate in un tipo DynamicFrame di dati di questo tipo e ne restituisce uno nuovo semplificato. DynamicFrame Se ci sono più tipi o un tipo di mappa contenuto in un tipo di elenco, gli elementi nell'elenco non verranno semplificati. Questo metodo supporta solo i dati nel formato JSON di esportazione DynamoDB. Prendi in considerazione unnest la possibilità di apportare modifiche simili su altri tipi di dati.

def simplifyDDBJson() : DynamicFrame 

Questo metodo non accetta alcun parametro.

Input di esempio

Prendi in considerazione lo schema seguente generato da un'esportazione DynamoDB:

root
|-- Item: struct
|    |-- parentMap: struct
|    |    |-- M: struct
|    |    |    |-- childMap: struct
|    |    |    |    |-- M: struct
|    |    |    |    |    |-- appName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- packageName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- updatedAt: struct
|    |    |    |    |    |    |-- N: string
|    |-- strings: struct
|    |    |-- SS: array
|    |    |    |-- element: string
|    |-- numbers: struct
|    |    |-- NS: array
|    |    |    |-- element: string
|    |-- binaries: struct
|    |    |-- BS: array
|    |    |    |-- element: string
|    |-- isDDBJson: struct
|    |    |-- BOOL: boolean
|    |-- nullValue: struct
|    |    |-- NULL: boolean

Esempio di codice

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContextimport scala.collection.JavaConverters._

object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.export" -> "ddb",
        "dynamodb.tableArn" -> "ddbTableARN",
        "dynamodb.s3.bucket" -> "exportBucketLocation",
        "dynamodb.s3.prefix" -> "exportBucketPrefix",
        "dynamodb.s3.bucketOwner" -> "exportBucketAccountID",
      ))
    ).getDynamicFrame()
    
    val simplified = dynamicFrame.simplifyDDBJson()
    simplified.printSchema()

    Job.commit()
  }

}

root
|-- parentMap: struct
|    |-- childMap: struct
|    |    |-- appName: string
|    |    |-- packageName: string
|    |    |-- updatedAt: string
|-- strings: array
|    |-- element: string
|-- numbers: array
|    |-- element: string
|-- binaries: array
|    |-- element: string
|-- isDDBJson: boolean
|-- nullValue: null

Def spigot

def spigot( path : String,
            options : JsonOptions = new JsonOptions("{}"),
            transformationContext : String = "",
            callSite : CallSite = CallSite("Not provided"),
            stageThreshold : Long = 0,
            totalThreshold : Long = 0
          ) : DynamicFrame 

Trasformazione passthrough che restituisce gli stessi record, ma scrive un sottoinsieme di record come effetto secondario.

Restituisce un oggetto DynamicFrame contenente gli stessi record di questo.

Per impostazione predefinita, scrive 100 record arbitrari nel percorso specificato da path. Questo comportamento può essere personalizzato usando la mappa options. Le chiavi valide includono le seguenti:

Ad esempio, la chiamata seguente esegue il campionamento del set di dati selezionando ogni record con una probabilità del 20% e arrestandosi dopo la scrittura di 200 record.

{{{
  df.spigot("s3://my_bucket/my_path", JsonOptions(Map("topk" -> 200, "prob" -> 0.2)))
}}}

Def splitFields

def splitFields( paths : Seq[String],
                 transformationContext : String = "",
                 callSite : CallSite = CallSite("Not provided", ""),
                 stageThreshold : Long = 0,
                 totalThreshold : Long = 0
               ) : Seq[DynamicFrame] 

Restituisce una sequenza di due oggetti DynamicFrame. Il primo oggetto DynamicFrame contiene i percorsi specificati, mentre il secondo contiene tutte le altre colonne.

Esempio

Questo esempio prende una tabella DynamicFrame creata dalla persons tabella nel legislators database del AWS Glue Data Catalog e la DynamicFrame divide in due, con i campi specificati che entrano nel primo DynamicFrame e i campi rimanenti in un secondo DynamicFrame. L'esempio sceglie quindi il primo DynamicFrame dal risultato.

val InputFrame = glueContext.getCatalogSource(database="legislators", tableName="persons", 
transformationContext="InputFrame").getDynamicFrame()

val SplitField_collection = InputFrame.splitFields(paths=Seq("family_name", "name", "links.note", 
"links.url", "gender", "image", "identifiers.scheme", "identifiers.identifier", "other_names.lang", 
"other_names.note", "other_names.name"), transformationContext="SplitField_collection")

val ResultFrame = SplitField_collection(0)

Def splitRows

def splitRows( paths : Seq[String],
               values : Seq[Any],
               operators : Seq[String],
               transformationContext : String,
               callSite : CallSite,
               stageThreshold : Long,
               totalThreshold : Long
             ) : Seq[DynamicFrame] 

Suddivide le righe in base a predicati che confrontano colonne e costanti.

Restituisce una sequenza di due oggetti DynamicFrame. Il primo contiene le righe per cui il predicato è true, il secondo contiene quelle per cui è false.

I predicati vengono specificati usando tre sequenze: "paths" contiene i nomi delle colonne (possibilmente annidate), "values" contiene i valori di costante rispetto ai quali eseguire il confronto e "operators" contiene gli operatori da usare per il confronto. Le tre sequenze devono essere tutte della stessa lunghezza: l'nesimo operatore verrà usato per confrontare la nesima colonna con l'nesimo valore.

Gli operatori consentiti sono: "!=", "=", "<=", "<", ">=" o ">".

Ad esempio, la chiamata seguente divide un oggetto DynamicFrame in modo che il primo frame di output contenga i record di persone degli Stati Uniti di età maggiore di 65 anni e che il secondo contenga tutti gli altri record.

{{{
  df.splitRows(Seq("age", "address.country"), Seq(65, "USA"), Seq(">=", "="))
}}}

Def stageErrorsCount

def stageErrorsCount

Restituisce il numero di record di errore creati durante il calcolo di questo oggetto DynamicFrame. Sono esclusi gli errori restituiti dalla operazioni precedenti passate a questo oggetto DynamicFrame come input.

Def toDF

def toDF( specs : Seq[ResolveSpec] = Seq.empty[ResolveSpec] ) : DataFrame 

Converte questo DynamicFrame in un Apache Spark SQL DataFrame con lo stesso schema e gli stessi record.

Def unbox

def unbox( path : String,
           format : String,
           optionString : String = "{}",
           transformationContext : String = "",
           callSite : CallSite = CallSite("Not provided"),
           stageThreshold : Long = 0,
           totalThreshold : Long = 0
         ) : DynamicFrame 

Analizza una stringa incorporata o una colonna binaria in base al formato specificato. Le colonne analizzate vengono annidate all'interno di una struttura con il nome di colonna originale.

Supponi, ad esempio, di avere un file CSV con una colonna JSON incorporata.

name, age, address
Sally, 36, {"state": "NE", "city": "Omaha"}
...

Dopo un'analisi iniziale, avresti un oggetto DynamicFrame con lo schema seguente.

{{{
  root
  |-- name: string
  |-- age: int
  |-- address: string
}}}

Puoi chiamare unbox nella colonna address per analizzare i componenti specifici.

{{{
  df.unbox("address", "json")
}}}

Otterrai un oggetto DynamicFrame con lo schema seguente.

{{{
  root
  |-- name: string
  |-- age: int
  |-- address: struct
  |    |-- state: string
  |    |-- city: string
}}}

Def unnest

def unnest( transformationContext : String = "",
            callSite : CallSite = CallSite("Not Provided"),
            stageThreshold : Long = 0,
            totalThreshold : Long = 0
          ) : DynamicFrame 

Restituisce un nuovo oggetto DynamicFrame con tutte le strutture annidate appiattite. I nomi vengono creati usando il carattere "." (punto).

Ad esempio, supponiamo di avere un DynamicFrame con lo schema seguente.

{{{
  root
  |-- name: string
  |-- age: int
  |-- address: struct
  |    |-- state: string
  |    |-- city: string
}}}

La chiamata seguente annulla l'annidamento della struttura address.

{{{
  df.unnest()
}}}

Lo schema risultante è il seguente.

{{{
  root
  |-- name: string
  |-- age: int
  |-- address.state: string
  |-- address.city: string
}}}

Questo metodo, inoltre, annulla l'annidamento delle strutture all'interno di array. Tuttavia, per motivi storici, ai nomi di tali campi vengono anteposti il nome della matrice di chiusura e ".val".

Def unnestDDBJson

unnestDDBJson(transformationContext : String = "",
            callSite : CallSite = CallSite("Not Provided"),
            stageThreshold : Long = 0,
            totalThreshold : Long = 0): DynamicFrame

Snidifica le colonne nidificate in un DynamicFrame che si trovano specificamente nella struttura JSON di DynamoDB e restituisce un nuovo DynamicFrame non annidato. Le colonne che sono di un array di struct non verranno annidate. Si noti che si tratta di un tipo specifico di trasformazione di snidamento che si comporta in modo diverso dalla normale trasformazione di unnest e richiede che i dati siano già nella struttura JSON di DynamoDB. Per ulteriori informazioni, consulta DynamoDB JSON.

Ad esempio, lo schema di lettura di un'esportazione con la struttura JSON DynamoDB potrebbe apparire come segue:

root
|-- Item: struct
|    |-- ColA: struct
|    |    |-- S: string
|    |-- ColB: struct
|    |    |-- S: string
|    |-- ColC: struct
|    |    |-- N: string
|    |-- ColD: struct
|    |    |-- L: array
|    |    |    |-- element: null

La trasformazione di unnestDDBJson() lo convertirebbe in:

root
|-- ColA: string
|-- ColB: string
|-- ColC: string
|-- ColD: array    
|    |-- element: null

L'esempio di codice seguente mostra come utilizzare il connettore di esportazione DynamoDB AWS Glue, richiamare un JSON di DynamoDB unnest e stampare il numero di partizioni:

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.export" -> "ddb",
        "dynamodb.tableArn" -> "<test_source>",
        "dynamodb.s3.bucket" -> "<bucket name>",
        "dynamodb.s3.prefix" -> "<bucket prefix>",
        "dynamodb.s3.bucketOwner" -> "<account_id of bucket>",
      ))
    ).getDynamicFrame()
    
    val unnested = dynamicFrame.unnestDDBJson()
    print(unnested.getNumPartitions())

    Job.commit()
  }

}

Def withFrameSchema

def withFrameSchema( getSchema : () => Schema ) : DynamicFrame 

Imposta lo schema di questo oggetto DynamicFrame sul valore specificato. Viene usato per lo più internamente per evitare ricalcoli dello schema onerosi. Lo schema passato deve contenere tutte le colonne presenti nei dati.

Def withName

def withName( name : String ) : DynamicFrame 

Restituisce una copia di questo oggetto DynamicFrame con un nuovo nome.

Def withTransformationContext

def withTransformationContext( ctx : String ) : DynamicFrame 

Restituisce una copia di questo oggetto DynamicFrame con il contesto di trasformazione specificato.