本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

AWS GlueScala API GlueContext

程序包：com.amazonaws.services.glue

class GlueContext extends SQLContext(sc) (
           @transient val sc : SparkContext,
           val defaultSourcePartitioner : PartitioningStrategy )

GlueContext 是在以下位置读取和写入 DynamicFrame 的入口点：Amazon Simple Storage Service（Amazon S3）、 AWS Glue 数据目录、JDBC 等。此类提供实用程序函数，用于创建可用来读取和写入 DynamicFrame 的 数据源特性 和 DataSink 对象。

如果从源创建的分区数小于分区的最小阈值（默认值为 10），您还可以使用 GlueContext 设置 DynamicFrame 中的目标分区数量（默认值为 20）。

def 列 addIngestionTime

def addIngestionTimeColumns(
         df : DataFrame, 
         timeGranularity : String = "") : dataFrame

将提取时间列（例如 ingest_year、ingest_month、ingest_day、ingest_hour、ingest_minute）附加到输入 DataFrame。当您指定以 Amazon S3 为目标的数据目录表时，AWS Glue 生成的脚本中会自动生成此函数。此函数使用输出表上的提取时间列自动更新分区。这允许根据提取时间自动对输出数据进行分区，而不需要在输入数据中显示提取时间列。

在附加时间粒度列后返回数据框。

例如：

glueContext.addIngestionTimeColumns(dataFrame, "hour")

def createDataFrame FromOptions

def createDataFrameFromOptions( connectionType : String,
                         connectionOptions : JsonOptions,
                         transformationContext : String = "",
                         format : String = null,
                         formatOptions : JsonOptions = JsonOptions.empty
                       ) : DataSource 

返回一个使用指定连接和格式创建的 DataFrame。此功能只能用于 AWS Glue 直播源。

Amazon Kinesis 流式传输源示例：

val data_frame_datasource0 = 
glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kinesis", 
connectionOptions = JsonOptions("""{"streamName": "example_stream", "startingPosition": "TRIM_HORIZON", "inferSchema": "true", "classification": "json"}}"""))

Kafka 流式传输源示例：

val data_frame_datasource0 = 
glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kafka", 
connectionOptions = JsonOptions("""{"connectionName": "example_connection", "topicName": "example_topic", "startingPosition": "earliest", "inferSchema": "false", "classification": "json", "schema":"`column1` STRING, `column2` STRING"}"""))

forEachBatch

示例：

glueContext.forEachBatch(data_frame_datasource0, (dataFrame: Dataset[Row], batchId: Long) => 
   {
      if (dataFrame.count() > 0) 
        {
          val datasource0 = DynamicFrame(glueContext.addIngestionTimeColumns(dataFrame, "hour"), glueContext)
          // @type: DataSink
          // @args: [database = "tempdb", table_name = "fromoptionsoutput", stream_batch_time = "100 seconds", 
          //      stream_checkpoint_location = "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/", 
          //      transformation_ctx = "datasink1"]
          // @return: datasink1
          // @inputs: [frame = datasource0]
          val options_datasink1 = JsonOptions(
             Map("partitionKeys" -> Seq("ingest_year", "ingest_month","ingest_day", "ingest_hour"), 
             "enableUpdateCatalog" -> true))
          val datasink1 = glueContext.getCatalogSink(
             database = "tempdb", 
             tableName = "fromoptionsoutput", 
             redshiftTmpDir = "", 
             transformationContext = "datasink1", 
             additionalOptions = options_datasink1).writeDynamicFrame(datasource0)
        }
   }, JsonOptions("""{"windowSize" : "100 seconds", 
         "checkpointLocation" : "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/"}"""))

def getCatalogSink

def getCatalogSink( database : String,
        tableName : String,
        redshiftTmpDir : String = "",
        transformationContext : String = ""
        additionalOptions: JsonOptions = JsonOptions.empty,
        catalogId: String = null   
) : DataSink 

创建一个可向在数据目录中定义的表中指定的位置写入的 DataSink。

返回 DataSink。

def getCatalogSource

def getCatalogSource( database : String,
                      tableName : String,
                      redshiftTmpDir : String = "",
                      transformationContext : String = ""
                      pushDownPredicate : String = " "
                      additionalOptions: JsonOptions = JsonOptions.empty,
                      catalogId: String = null
                    ) : DataSource 

创建可从数据目录中的表定义读取数据的 数据源特性。

返回 DataSource。

流式传输源示例

val data_frame_datasource0 = glueContext.getCatalogSource(
    database = "tempdb",
    tableName = "test-stream-input", 
    redshiftTmpDir = "", 
    transformationContext = "datasource0", 
    additionalOptions = JsonOptions("""{
        "startingPosition": "TRIM_HORIZON", "inferSchema": "false"}""")
    ).getDataFrame()

def getJDBCSink

def getJDBCSink( catalogConnection : String,
                 options : JsonOptions,
                 redshiftTmpDir : String = "",
                 transformationContext : String = "",
                 catalogId: String = null
               ) : DataSink 

创建一个可向在数据目录中的 Connection 对象指定的 JDBC 数据库写入的 DataSink。Connection 对象具有用于连接 JDBC 接收器的信息，包括 URL、用户名、密码、VPC、子网和安全组。

示例代码：

getJDBCSink(catalogConnection = "my-connection-name", options = JsonOptions("""{"dbtable": "my-jdbc-table", "database": "my-jdbc-db"}"""), redshiftTmpDir = "", transformationContext = "datasink4")

返回 DataSink。

def getSink

def getSink( connectionType : String,
             connectionOptions : JsonOptions,
             transformationContext : String = ""
           ) : DataSink 

创建将数据写入目标的DataSink，例如亚马逊简单存储服务 (Amazon S3)、JDBC 或 Glue 数据目录，或者是 AWS Apache Kafka 或 Amazon Kinesis 数据流。

返回 DataSink。

def 格式 getSinkWith

def getSinkWithFormat( connectionType : String,
                       options : JsonOptions,
                       transformationContext : String = "",
                       format : String = null,
                       formatOptions : JsonOptions = JsonOptions.empty
                     ) : DataSink 

创建 DataSink，将数据写入 Amazon S3、JDBC、数据目录、Apache Kafka 或 Amazon Kinesis 数据流等目标写入数据。此外还将设置要写出到目标的数据格式。

返回 DataSink。

def getSource

def getSource( connectionType : String,
               connectionOptions : JsonOptions,
               transformationContext : String = ""
               pushDownPredicate
             ) : DataSource 

创建从 Amazon S3、JDBC 或 Glue 数据目录等来源读取数据 AWS 的。数据源特性还支持 Kafka 和 Kinesis 流式传输数据源。

返回 DataSource。

Amazon Kinesis 流式传输源示例：

val kinesisOptions = jsonOptions()
data_frame_datasource0 = glueContext.getSource("kinesis", kinesisOptions).getDataFrame()

private def jsonOptions(): JsonOptions = {
    new JsonOptions(
      s"""{"streamARN": "arn:aws:kinesis:eu-central-1:123456789012:stream/fromOptionsStream",
         |"startingPosition": "TRIM_HORIZON",
         |"inferSchema": "true",
         |"classification": "json"}""".stripMargin)
}

Kafka 流式传输源示例：

val kafkaOptions = jsonOptions()
val data_frame_datasource0 = glueContext.getSource("kafka", kafkaOptions).getDataFrame()

private def jsonOptions(): JsonOptions = {
    new JsonOptions(
      s"""{"connectionName": "ConfluentKafka",
         |"topicName": "kafka-auth-topic",
         |"startingOffsets": "earliest",
         |"inferSchema": "true",
         |"classification": "json"}""".stripMargin)
 }

def 格式 getSourceWith

def getSourceWithFormat( connectionType : String,
                         options : JsonOptions,
                         transformationContext : String = "",
                         format : String = null,
                         formatOptions : JsonOptions = JsonOptions.empty
                       ) : DataSource 

创建一个数据源特性，用于从 Amazon S3、JDBC 或 Glu AWS e 数据目录等来源读取数据，并设置存储在源中的数据的格式。

返回 DataSource。

示例

在 A DynamicFrame mazon S3 上使用逗号分隔值 (CSV) 文件的数据源创建：

val datasource0 = glueContext.getSourceWithFormat(
    connectionType="s3",
    options =JsonOptions(s"""{"paths": [ "s3://csv/nycflights.csv"]}"""),
    transformationContext = "datasource0", 
    format = "csv",
    formatOptions=JsonOptions(s"""{"withHeader":"true","separator": ","}""")
    ).getDynamicFrame()

使用 JDBC 连接 DynamicFrame 从 PostgreSQL 数据源创建：

val datasource0 = glueContext.getSourceWithFormat(
    connectionType="postgresql",
    options =JsonOptions(s"""{
      "url":"jdbc:postgresql://databasePostgres-1.rds.amazonaws.com:5432/testdb",
      "dbtable": "public.company",
      "redshiftTmpDir":"", 
      "user":"username", 
      "password":"password123"
    }"""),
    transformationContext = "datasource0").getDynamicFrame()

使用 JDBC 连接 DynamicFrame 从 MySQL 的数据源创建：

 val datasource0 = glueContext.getSourceWithFormat(
    connectionType="mysql",
    options =JsonOptions(s"""{
      "url":"jdbc:mysql://databaseMysql-1.rds.amazonaws.com:3306/testdb",
      "dbtable": "athenatest_nycflights13_csv",
      "redshiftTmpDir":"", 
      "user":"username", 
      "password":"password123"
    }"""),
    transformationContext = "datasource0").getDynamicFrame()

def getSparkSession

def getSparkSession : SparkSession 

获取与此 GlueContext 关联的 SparkSession 对象。使用此 SparkSession 对象注册表和 UDF，以便在DataFrame创建时 DynamicFrames使用。

返回 SparkSession。

def startTransaction

def startTransaction(readOnly: Boolean):String

开启新事务。内部调用 Lake Formation startTransaction API。

返回事务 ID。

def commitTransaction

def commitTransaction(transactionId: String, waitForCommit: Boolean): Boolean

尝试提交指定的事务。可能在事务完成提交之前返回 commitTransaction。内部调用 Lake Formation commitTransaction API。

返回布尔值，以指示是否完成提交。

def cancelTransaction

def cancelTransaction(transactionId: String): Unit

尝试取消指定的事务。在内部调用 Lake Format CancelTransactionion API。

如果事务以前已提交，则返回 TransactionCommittedException 异常。

def this

def this( sc : SparkContext,
          minPartitions : Int,
          targetPartitions : Int )

使用指定的 SparkContext、最小分区数和目标分区数创建 GlueContext 对象。

返回 GlueContext。

def this

def this( sc : SparkContext )

使用提供的 GlueContext 创建一个 SparkContext 对象。将最小分区数设置为 10，将目标分区数设置为 20。

返回 GlueContext。

def this

def this( sparkContext : JavaSparkContext )

使用提供的 GlueContext 创建一个 JavaSparkContext 对象。将最小分区数设置为 10，将目标分区数设置为 20。

返回 GlueContext。