连接到外部系统

连接到外部系统

译者：flink.sojb.cn

Flink的 Table API和SQL程序可以连接到其他外部系统，以便读取和写入批处理表和流表。表源提供对存储在外部系统（例如数据库，键值存储，消息队列或文件系统）中的数据的访问。表接收器向外部存储系统发出表。根据源和接收器的类型，它们支持不同的格式，如CSV，Parquet或ORC。

本页介绍如何声明内置表源和/或表接收器，并在Flink中注册它们。注册源或接收器后，可以通过 Table API和SQL语句访问它。

注意如果要实现自己的自定义表源或接收器，请查看用户定义的源和接收器页面。

依赖

下表列出了所有可用的连接器和格式。它们的相互兼容性在表连接器和表格格式的相应部分中标记。下表提供了使用构建自动化工具（如Maven或SBT）和带有SQL JAR包的SQL Client的两个项目的依赖关系信息。

此表仅适用于稳定版本。

概览

从Flink 1.6开始，与外部系统的连接声明与实际实现分开。

也可以指定连接

以编程方式使用Descriptorunder org.apache.flink.table.descriptorsfor Table＆SQL API
或声明性地通过SQL客户端的YAML配置文件。

这不仅可以更好地统一API和SQL Client，还可以在自定义实现的情况下实现更好的可扩展性，而无需更改实际声明。

每个声明都类似于SQL CREATE TABLE语句。可以预先定义表的名称，表的模式，连接器以及用于连接到外部系统的数据格式。

的连接器描述了存储表的数据的外部系统。可以在此处声明Apacha Kafka或常规文件系统等存储系统。连接器可能已经提供了具有字段和架构的固定格式。

某些系统支持不同的数据格式。例如，存储在Kafka或文件中的表可以使用CSV，JSON或Avro对其行进行编码。数据库连接器可能需要此处的表模式。每个连接器都记录了存储系统是否需要定义格式。不同的系统还需要不同类型的格式（例如，面向列的格式与面向行的格式）。该文档说明了哪些格式类型和连接器兼容。

该表架构定义了暴露在SQL查询表的架构。它描述了源如何将数据格式映射到表模式，反之亦然。模式可以访问连接器或格式定义的字段。它可以使用一个或多个字段来提取或插入时间属性。如果输入字段没有确定的字段顺序，则架构清楚地定义列名称，它们的顺序和原点。

后续部分将更详细地介绍每个定义部分（连接器，格式和架构）。以下示例显示了如何传递它们：

Java/Scala
YAML

tableEnvironment
  .connect(...)
  .withFormat(...)
  .withSchema(...)
  .inAppendMode()
  .registerTableSource("MyTable")

name: MyTable
type: source
update-mode: append
connector: ...
format: ...
schema: ...

表的类型（source，sink或both）确定表的注册方式。在表类型的情况下，表both源和表接收器都以相同的名称注册。从逻辑上讲，这意味着我们可以读取和写入这样的表，类似于常规DBMS中的表。

对于流式查询，更新模式声明如何在动态表和存储系统之间进行通信以进行连续查询。

以下代码显示了如何连接到Kafka以读取Avro记录的完整示例。

Java/Scala
YAML

tableEnvironment
  // declare the external system to connect to
  .connect(
    new Kafka()
      .version("0.10")
      .topic("test-input")
      .startFromEarliest()
      .property("zookeeper.connect", "localhost:2181")
      .property("bootstrap.servers", "localhost:9092")
  )
  // declare a format for this system
  .withFormat(
    new Avro()
      .avroSchema(
        "{" +
        "  \"namespace\": \"org.myorganization\"," +
        "  \"type\": \"record\"," +
        "  \"name\": \"UserMessage\"," +
        "    \"fields\": [" +
        "      {\"name\": \"timestamp\", \"type\": \"string\"}," +
        "      {\"name\": \"user\", \"type\": \"long\"}," +
        "      {\"name\": \"message\", \"type\": [\"string\", \"null\"]}" +
        "    ]" +
        "}" +
      )
  )
  // declare the schema of the table
  .withSchema(
    new Schema()
      .field("rowtime", Types.SQL_TIMESTAMP)
        .rowtime(new Rowtime()
          .timestampsFromField("ts")
          .watermarksPeriodicBounded(60000)
        )
      .field("user", Types.LONG)
      .field("message", Types.STRING)
  )
  // specify the update-mode for streaming tables
  .inAppendMode()
  // register as source, sink, or both and under a name
  .registerTableSource("MyUserTable");

tables:
  - name: MyUserTable      # name the new table
    type: source           # declare if the table should be "source", "sink", or "both"
    update-mode: append    # specify the update-mode for streaming tables
    # declare the external system to connect to
    connector:
      type: kafka
      version: "0.10"
      topic: test-input
      startup-mode: earliest-offset
      properties:
        - key: zookeeper.connect
          value: localhost:2181
        - key: bootstrap.servers
          value: localhost:9092
    # declare a format for this system
    format:
      type: avro
      avro-schema: >
        {
          "namespace": "org.myorganization",
          "type": "record",
          "name": "UserMessage",
            "fields": [
              {"name": "ts", "type": "string"},
              {"name": "user", "type": "long"},
              {"name": "message", "type": ["string", "null"]}
            ]
        }
    # declare the schema of the table
    schema:
      - name: rowtime
        type: TIMESTAMP
        rowtime:
          timestamps:
            type: from-field
            from: ts
          watermarks:
            type: periodic-bounded
            delay: "60000"
      - name: user
        type: BIGINT
      - name: message
        type: VARCHAR

在两种方式中，所需的连接属性都转换为规范化的，基于字符串的键值对。所谓的表工厂从键值对创建配置的表源，表接收器和相应的格式。在搜索完全匹配的表工厂时，会考虑通过Java服务提供程序接口（SPI）找到的所有表工厂。

如果找不到工厂或多个工厂匹配给定的属性，则会抛出一个异常，其中包含有关已考虑的工厂和支持的属性的其他信息。

表格式

表模式定义类的名称和类型，类似于SQL CREATE TABLE语句的列定义。另外，可以指定列的表示方式和表格数据编码格式的字段。如果列的名称应与输入/输出格式不同，则字段的来源可能很重要。例如，列user_name应从$$-user-nameJSON格式引用该字段。此外，需要架构将类型从外部系统映射到Flink的表示。对于表接收器，它确保仅将具有有效模式的数据写入外部系统。

以下示例显示了一个没有时间属性的简单模式，以及输入/输出到表列的一对一字段映射。

Java/Scala
YAML

.withSchema(
  new Schema()
    .field("MyField1", Types.INT)     // required: specify the fields of the table (in this order)
    .field("MyField2", Types.STRING)
    .field("MyField3", Types.BOOLEAN)
)

schema:
  - name: MyField1    # required: specify the fields of the table (in this order)
    type: INT
  - name: MyField2
    type: VARCHAR
  - name: MyField3
    type: BOOLEAN

对于每个字段，除了列的名称和类型之外，还可以声明以下属性：

Java/Scala
YAML

.withSchema(
  new Schema()
    .field("MyField1", Types.SQL_TIMESTAMP)
      .proctime()      // optional: declares this field as a processing-time attribute
    .field("MyField2", Types.SQL_TIMESTAMP)
      .rowtime(...)    // optional: declares this field as a event-time attribute
    .field("MyField3", Types.BOOLEAN)
      .from("mf3")     // optional: original field in the input that is referenced/aliased by this field
)

schema:
  - name: MyField1
    type: TIMESTAMP
    proctime: true    # optional: boolean flag whether this field should be a processing-time attribute
  - name: MyField2
    type: TIMESTAMP
    rowtime: ...      # optional: wether this field should be a event-time attribute
  - name: MyField3
    type: BOOLEAN
    from: mf3         # optional: original field in the input that is referenced/aliased by this field

使用无界流表时，时间属性是必不可少的。因此，处理时间和事件时间（也称为“行时”）属性都可以定义为模式的一部分。

有关Flink中时间处理的更多信息，特别是事件时间，我们建议使用常规事件时间部分。

行时属性

为了控制表的事件时间行为，Flink提供了预定义的时间戳提取器和水印策略。

支持以下时间戳提取器：

Java/Scala
YAML

// Converts an existing LONG or SQL_TIMESTAMP field in the input into the rowtime attribute.
.rowtime(
  new Rowtime()
    .timestampsFromField("ts_field")    // required: original field name in the input
)
// Converts the assigned timestamps from a DataStream API record into the rowtime attribute 
// and thus preserves the assigned timestamps from the source.
// This requires a source that assigns timestamps (e.g., Kafka 0.10+).
.rowtime(
  new Rowtime()
    .timestampsFromSource()
)
// Sets a custom timestamp extractor to be used for the rowtime attribute.
// The extractor must extend `org.apache.flink.table.sources.tsextractors.TimestampExtractor`.
.rowtime(
  new Rowtime()
    .timestampsFromExtractor(...)
)

# Converts an existing BIGINT or TIMESTAMP field in the input into the rowtime attribute.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
rowtime:
  timestamps:
    type: from-field
    from: "ts_field"                 # required: original field name in the input
# Converts the assigned timestamps from a DataStream API record into the rowtime attribute 
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
# and thus preserves the assigned timestamps from the source.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
rowtime:
  timestamps:
    type: from-source

支持以下水印策略：

Java/Scala
YAML

// Sets a watermark strategy for ascending rowtime attributes. Emits a watermark of the maximum 
// observed timestamp so far minus 1\. Rows that have a timestamp equal to the max timestamp
// are not late.
.rowtime(
  new Rowtime()
    .watermarksPeriodicAscending()
)
// Sets a built-in watermark strategy for rowtime attributes which are out-of-order by a bounded time interval.
// Emits watermarks which are the maximum observed timestamp minus the specified delay.
.rowtime(
  new Rowtime()
    .watermarksPeriodicBounded(2000)    // delay in milliseconds
)
// Sets a built-in watermark strategy which indicates the watermarks should be preserved from the
// underlying DataStream API and thus preserves the assigned watermarks from the source.
.rowtime(
  new Rowtime()
    .watermarksFromSource()
)

# Sets a watermark strategy for ascending rowtime attributes. Emits a watermark of the maximum 
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
# observed timestamp so far minus 1\. Rows that have a timestamp equal to the max timestamp
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
# are not late.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
rowtime:
  watermarks:
    type: periodic-ascending
# Sets a built-in watermark strategy for rowtime attributes which are out-of-order by a bounded time interval.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
# Emits watermarks which are the maximum observed timestamp minus the specified delay.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
rowtime:
  watermarks:
    type: periodic-bounded
    delay: ...                # required: delay in milliseconds
# Sets a built-in watermark strategy which indicates the watermarks should be preserved from the
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
# underlying DataStream API and thus preserves the assigned watermarks from the source.
> 译者：[flink.sojb.cn](https://flink.sojb.cn/)
rowtime:
  watermarks:
    type: from-source

确保始终声明时间戳和水印。触发基于时间的算子操作需要水印。

类型字符串

由于类型信息仅在编程语言中可用，因此支持在YAML文件中定义以下类型字符串：

VARCHAR
BOOLEAN
TINYINT
SMALLINT
INT
BIGINT
FLOAT
DOUBLE
DECIMAL
DATE
TIME
TIMESTAMP
ROW(fieldtype, ...)              # unnamed row; e.g. ROW(VARCHAR, INT) that is mapped to Flink's RowTypeInfo
                                 # with indexed fields names f0, f1, ...
ROW(fieldname fieldtype, ...)    # named row; e.g., ROW(myField VARCHAR, myOtherField INT) that
                                 # is mapped to Flink's RowTypeInfo
POJO(class)                      # e.g., POJO(org.mycompany.MyPojoClass) that is mapped to Flink's PojoTypeInfo
ANY(class)                       # e.g., ANY(org.mycompany.MyClass) that is mapped to Flink's GenericTypeInfo
ANY(class, serialized)           # used for type information that is not supported by Flink's Table & SQL API

更新模式

对于流式查询，需要声明如何在动态表和外部连接器之间执行转换。的更新模式指定哪些类型的消息应与外部系统进行交换：

追加模式：在追加模式下，动态表和外部连接器仅交换INSERT消息。

回退模式：在回退模式下，动态表和外部连接器交换ADD和RETRACT消息。INSERT更改被编码为ADD消息，DELETE更改被编码为RETRACT消息，UPDATE更改被编码为更新（先前）行的RETRACT消息和更新（新）行的ADD消息。在此模式下，不能定义键而不是upsert模式。但是，每次更新都包含两个效率较低的消息。

Upsert模式：在upsert模式下，动态表和外部连接器交换UPSERT和DELETE消息。此模式需要一个（可能是复合的）唯一键，通过该键可以传播更新。外部连接器需要知道唯一键属性才能正确应用消息。INSERT和UPDATE更改被编码为UPSERT消息。DELETE更改为DELETE消息。与收回流的主要区别在于UPDATE更改使用单个消息进行编码，因此更有效。

注意每个连接器的文档都说明支持哪些更新模式。

Java/Scala
YAML

.connect(...)
  .inAppendMode()    // otherwise: inUpsertMode() or inRetractMode()

tables:
  - name: ...
    update-mode: append    # otherwise: "retract" or "upsert"

有关更多信息，另请参阅常规流概念文档。

表连接器

Flink提供了一组用于连接外部系统的连接器。

请注意，并非所有连接器都可用于批量和流式传输。此外，并非每个流连接器都支持每种流模式。因此，相应地标记每个连接器。格式标记表示连接器需要特定类型的格式。

文件系统连接器

来源：批量来源：流处理附加模式接收器：批量接收器：流式附加模式格式：仅限CSV

文件系统连接器允许从本地或分布式文件系统进行读写。文件系统可以定义为：

Java/Scala
YAML

.connect(
  new FileSystem()
    .path("file:///path/to/whatever")    // required: path to a file or directory
)

connector:
  type: filesystem
  path: "file:///path/to/whatever"    # required: path to a file or directory

文件系统连接器本身包含在Flink中，不需要额外的依赖项。需要指定相应的格式，以便从文件系统读取和写入行。

注意确保包含特定于Flink文件系统的依赖项。

注意用于流式传输的文件系统源和接收器仅是实验性的。将来，我们将支持实际的流式使用案例，即目录监控和桶输出。

Kafka连接器

来源：流附加模式接收器：流附加模式格式：序列化模式格式：反序列化模式

Kafka连接器允许从Apache Kafka主题读取和写入。它可以定义如下：

Java/Scala
YAML

.connect(
  new Kafka()
    .version("0.11")    // required: valid connector versions are "0.8", "0.9", "0.10", and "0.11"
    .topic("...")       // required: topic name from which the table is read
    // optional: connector specific properties
    .property("zookeeper.connect", "localhost:2181")
    .property("bootstrap.servers", "localhost:9092")
    .property("group.id", "testGroup")
    // optional: select a startup mode for Kafka offsets
    .startFromEarliest()
    .startFromLatest()
    .startFromSpecificOffsets(...)
    // optional: output partitioning from Flink's partitions into Kafka's partitions
    .sinkPartitionerFixed()         // each Flink partition ends up in at-most one Kafka partition (default)
    .sinkPartitionerRoundRobin()    // a Flink partition is distributed to Kafka partitions round-robin
    .sinkPartitionerCustom(MyCustom.class)    // use a custom FlinkKafkaPartitioner subclass
)

connector:
  type: kafka
  version: 0.11       # required: valid connector versions are "0.8", "0.9", "0.10", and "0.11"
  topic: ...          # required: topic name from which the table is read
  properties:         # optional: connector specific properties
    - key: zookeeper.connect
      value: localhost:2181
    - key: bootstrap.servers
      value: localhost:9092
    - key: group.id
      value: testGroup
  startup-mode: ...   # optional: valid modes are "earliest-offset", "latest-offset",
                      # "group-offsets", or "specific-offsets"
  specific-offsets:   # optional: used in case of startup mode with specific offsets
    - partition: 0
      offset: 42
    - partition: 1
      offset: 300
  sink-partitioner: ...    # optional: output partitioning from Flink's partitions into Kafka's partitions
                           # valid are "fixed" (each Flink partition ends up in at most one Kafka partition),
                           # "round-robin" (a Flink partition is distributed to Kafka partitions round-robin)
                           # "custom" (use a custom FlinkKafkaPartitioner subclass)
  sink-partitioner-class: org.mycompany.MyPartitioner  # optional: used in case of sink partitioner custom

指定开始读取位置：默认情况下，Kafka源将开始从Zookeeper或Kafka代理中的已提交组偏移量中读取数据。您可以指定其他起始位置，这些位置对应于Kafka消费者起始位置配置部分中的配置。

Flink-Kafka接收器分区：默认情况下，Kafka接收器最多写入与其自身并行性一样多的分区（接收器的每个并行实例仅写入一个分区）。为了将写入分发到更多分区或控制行到分区的路由，可以提供自定义接收器分区器。循环分区器可用于避免不平衡的分区。但是，它会在所有Flink实例和所有Kafka代理之间产生大量网络连接。

一致性保证：默认情况下，如果在启用了检查点的情况下执行查询，则Kafka接收器会使用至少一次保证将数据提取到Kafka主题中。

Kafka 0.10+时间戳：自Kafka 0.10起，Kafka消息的时间戳作为元数据，指定记录何时写入Kafka主题。通过分别在YAML和Java / Scala中选择，可以将这些时间戳用于行时属性。timestamps: from-source``timestampsFromSource()

确保添加特定于版本的Kafka依赖项。此外，需要指定相应的格式以便从Kafka读取和写入行。

表格格式

Flink提供了一组可与表连接器一起使用的表格格式。

格式标记表示与连接器匹配的格式类型。

CSV格式

CSV格式允许读取和写入以逗号分隔的行。

Java/Scala
YAML

.withFormat(
  new Csv()
    .field("field1", Types.STRING)    // required: ordered format fields
    .field("field2", Types.TIMESTAMP)
    .fieldDelimiter(",")              // optional: string delimiter "," by default 
    .lineDelimiter("\n")              // optional: string delimiter "\n" by default 
    .quoteCharacter('"')              // optional: single character for string values, empty by default
    .commentPrefix('#')               // optional: string to indicate comments, empty by default
    .ignoreFirstLine()                // optional: ignore the first line, by default it is not skipped
    .ignoreParseErrors()              // optional: skip records with parse error instead of failing by default
)

format:
  type: csv
  fields:                    # required: ordered format fields
    - name: field1
      type: VARCHAR
    - name: field2
      type: TIMESTAMP
  field-delimiter: ","       # optional: string delimiter "," by default 
  line-delimiter: "\n"       # optional: string delimiter "\n" by default 
  quote-character: '"'       # optional: single character for string values, empty by default
  comment-prefix: '#'        # optional: string to indicate comments, empty by default
  ignore-first-line: false   # optional: boolean flag to ignore the first line, by default it is not skipped
  ignore-parse-errors: true  # optional: skip records with parse error instead of failing by default

CSV格式包含在Flink中，不需要其他依赖项。

注意目前写入行的CSV格式有限。仅支持自定义字段分隔符作为可选参数。

JSON格式

格式：序列化架构格式：反序列化架构

JSON格式允许读取和写入与给定格式模式相对应的JSON数据。格式模式可以定义为Flink类型，JSON模式，也可以从所需的表模式派生。Flink类型支持更类似SQL的定义并映射到相应的SQL数据类型。JSON模式允许更复杂和嵌套的结构。

如果格式架构等于表架构，则还可以自动派生架构。这允许仅定义一次架构信息。格式的名称，类型和字段顺序由表的架构决定。如果时间属性的来源不是字段，则会忽略它们。一个from表中的模式定义解释为格式字段命名。

Java/Scala
YAML

.withFormat(
  new Json()
    .failOnMissingField(true)   // optional: flag whether to fail if a field is missing or not, false by default
    // required: define the schema either by using type information which parses numbers to corresponding types
    .schema(Type.ROW(...))
    // or by using a JSON schema which parses to DECIMAL and TIMESTAMP
    .jsonSchema(
      "{" +
      "  type: 'object'," +
      "  properties: {" +
      "    lon: {" +
      "      type: 'number'" +
      "    }," +
      "    rideTime: {" +
      "      type: 'string'," +
      "      format: 'date-time'" +
      "    }" +
      "  }" +
      "}"
    )
    // or use the table's schema
    .deriveSchema()
)

format:
  type: json
  fail-on-missing-field: true   # optional: flag whether to fail if a field is missing or not, false by default
  # required: define the schema either by using a type string which parses numbers to corresponding types
  schema: "ROW(lon  FLOAT,  rideTime  TIMESTAMP)"
  # or by using a JSON schema which parses to DECIMAL and TIMESTAMP
  json-schema: >
    {
      type: 'object',
      properties: {
        lon: {
          type: 'number'
        },
        rideTime: {
          type: 'string',
          format: 'date-time'
        }
      }
    }
  # or use the table's schema
  derive-schema: true

下表显示了JSON模式类型到Flink SQL类型的映射：

JSON模式	Flink SQL
`object`	`ROW`
`boolean`	`BOOLEAN`
`array`	`ARRAY[_]`
`number`	`DECIMAL`
`integer`	`DECIMAL`
`string`	`VARCHAR`
`string` with `format: date-time`	`TIMESTAMP`
`string` with `format: date`	`DATE`
`string` with `format: time`	`TIME`
`string` with `encoding: base64`	`ARRAY[TINYINT]`
`null`	`NULL` （尚不支持）

目前，Flink仅支持JSON模式规范的子集draft-07。Union类型（以及allOf，anyOf，not）尚未支持。oneOf和类型数组仅支持指定可为空性。

支持链接到文档中的通用定义的简单引用，如下面更复杂的示例所示：

{  "definitions":  {  "address":  {  "type":  "object",  "properties":  {  "street_address":  {  "type":  "string"  },  "city":  {  "type":  "string"  },  "state":  {  "type":  "string"  }  },  "required":  [  "street_address",  "city",  "state"  ]  }  },  "type":  "object",  "properties":  {  "billing_address":  {  "$ref":  "#/definitions/address"  },  "shipping_address":  {  "$ref":  "#/definitions/address"  },  "optional_address":  {  "oneOf":  [  {  "type":  "null"  },  {  "$ref":  "#/definitions/address"  }  ]  }  }  }

缺少字段处理：默认情况下，缺少的JSON字段设置为null。您可以启用严格的JSON解析，如果缺少字段，将取消源（和查询）。

确保将JSON格式添加为依赖项。

Apache Avro格式

格式：序列化架构格式：反序列化架构

在Apache的Avro的格式允许读取和写入对应于给定的格式模式Avro的数据。格式模式可以定义为Avro特定记录的完全限定类名，也可以定义为Avro架构字符串。如果使用类名，则在运行时期间类必须在类路径中可用。

Java/Scala
YAML

.withFormat(
  new Avro()
    // required: define the schema either by using an Avro specific record class
    .recordClass(User.class)
    // or by using an Avro schema
    .avroSchema(
      "{" +
      "  \"type\": \"record\"," +
      "  \"name\": \"test\"," +
      "  \"fields\" : [" +
      "    {\"name\": \"a\", \"type\": \"long\"}," +
      "    {\"name\": \"b\", \"type\": \"string\"}" +
      "  ]" +
      "}"
    )
)

format:
  type: avro
  # required: define the schema either by using an Avro specific record class
  record-class: "org.organization.types.User"
  # or by using an Avro schema
  avro-schema: >
    {
      "type": "record",
      "name": "test",
      "fields" : [
        {"name": "a", "type": "long"},
        {"name": "b", "type": "string"}
      ]
    }

Avro类型映射到相应的SQL数据类型。仅支持联合类型以指定可为空性，否则它们将转换为ANY类型。下表显示了映射：

Avro架构	Flink SQL
`record`	`ROW`
`enum`	`VARCHAR`
`array`	`ARRAY[_]`
`map`	`MAP[VARCHAR, _]`
`union`	非null类型或 `ANY`
`fixed`	`ARRAY[TINYINT]`
`string`	`VARCHAR`
`bytes`	`ARRAY[TINYINT]`
`int`	`INT`
`long`	`BIGINT`
`float`	`FLOAT`
`double`	`DOUBLE`
`boolean`	`BOOLEAN`
`int` with `logicalType: date`	`DATE`
`int` with `logicalType: time-millis`	`TIME`
`int` with `logicalType: time-micros`	`INT`
`long` with `logicalType: timestamp-millis`	`TIMESTAMP`
`long` with `logicalType: timestamp-micros`	`BIGINT`
`bytes` with `logicalType: decimal`	`DECIMAL`
`fixed` with `logicalType: decimal`	`DECIMAL`
`null`	`NULL` （尚不支持）

Avro使用Joda-Time来表示特定记录类中的逻辑日期和时间类型。Joda-Time依赖不是Flink分发的一部分。因此，请确保Joda-Time在运行时期间与您的特定记录类一起位于类路径中。通过模式字符串指定的Avro格式不需要存在Joda-Time。

确保添加Apache Avro依赖项。

进一步的TableSources和TableSinks

尚未将以下表源和接收器迁移（或尚未完全迁移）到新的统一接口。

这些是TableSourceFlink提供的附加函数：

班级名称	Maven依赖	批量？	流？	描述
`OrcTableSource`	`flink-orc`	Y	N	一个`TableSource`ORC文件。

这些是TableSinkFlink提供的附加函数：

班级名称	Maven依赖	批量？	流？	描述
`CsvTableSink`	`flink-table`	Y	附加	CSV文件的简单接收器。
`JDBCAppendTableSink`	`flink-jdbc`	Y	附加	将表写入JDBC表。
`CassandraAppendTableSink`	`flink-connector-cassandra`	N	附加	将表写入Cassandra表。

OrcTableSource

在OrcTableSource读取ORC文件。ORC是结构化数据的文件格式，并以压缩的柱状表示形式存储数据。ORC非常高效，支持Projection和滤波器下推。

一个OrcTableSource被创建，如下所示：

Java
Scala

// create Hadoop Configuration
Configuration config = new Configuration();
OrcTableSource orcTableSource = OrcTableSource.builder()
  // path to ORC file(s). NOTE: By default, directories are recursively scanned.
  .path("file:///path/to/data")
  // schema of ORC files
  .forOrcSchema("struct<name:string,addresses:array<struct<street:string,zip:smallint>>>")
  // Hadoop configuration
  .withConfiguration(config)
  // build OrcTableSource
  .build();

// create Hadoop Configuration val config = new Configuration()
val orcTableSource = OrcTableSource.builder()
  // path to ORC file(s). NOTE: By default, directories are recursively scanned.
  .path("file:///path/to/data")
  // schema of ORC files
  .forOrcSchema("struct<name:string,addresses:array<struct<street:string,zip:smallint>>>")
  // Hadoop configuration
  .withConfiguration(config)
  // build OrcTableSource
  .build()

注：在OrcTableSource不支持ORC的Union类型呢。

CsvTableSink

在CsvTableSink发出Table一个或一个以上的CSV文件。

接收器仅支持仅附加流表。它不能用于发出Table不断更新的内容。有关详细信息，请参阅表到流转换的文档。发出流表时，行至少写入一次（如果启用了检查点），CsvTableSink并且不将输出文件拆分为存储桶文件，而是连续写入相同的文件。

Java
Scala

Table table = ...
table.writeToSink(
  new CsvTableSink(
    path,                  // output path 
    "|",                   // optional: delimit files by '|'
    1,                     // optional: write to a single file
    WriteMode.OVERWRITE)); // optional: override existing files

val table: Table = ???
table.writeToSink(
  new CsvTableSink(
    path,                             // output path
    fieldDelim = "|",                 // optional: delimit files by '|'
    numFiles = 1,                     // optional: write to a single file
    writeMode = WriteMode.OVERWRITE)) // optional: override existing files

JDBCAppendTableSink

在JDBCAppendTableSink发出Table一个JDBC连接。接收器仅支持仅附加流表。它不能用于发出Table不断更新的内容。有关详细信息，请参阅表到流转换的文档。

所述JDBCAppendTableSink插入物每Table行至少一次到数据库表（如果启用了检查点）。但是，您可以使用REPLACE或指定插入查询来指定INSERT OVERWRITE对数据库的写入。

要使用JDBC接收器，必须将JDBC连接器依赖项（flink-jdbc）添加到项目中。然后您可以使用JDBCAppendSinkBuilder以下方法创建接收器

Java
Scala

JDBCAppendTableSink sink = JDBCAppendTableSink.builder()
  .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
  .setDBUrl("jdbc:derby:memory:ebookshop")
  .setQuery("INSERT INTO books (id) VALUES (?)")
  .setParameterTypes(INT_TYPE_INFO)
  .build();
Table table = ...
table.writeToSink(sink);

val sink: JDBCAppendTableSink = JDBCAppendTableSink.builder()
  .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
  .setDBUrl("jdbc:derby:memory:ebookshop")
  .setQuery("INSERT INTO books (id) VALUES (?)")
  .setParameterTypes(INT_TYPE_INFO)
  .build()
val table: Table = ???
table.writeToSink(sink)

与using类似JDBCOutputFormat，您必须显式指定JDBC驱动程序的名称，JDBC URL，要执行的查询以及JDBC表的字段类型。

CassandraAppendTableSink

该CassandraAppendTableSink发射Table到卡桑德拉表。接收器仅支持仅附加流表。它不能用于发出Table不断更新的内容。有关详细信息，请参阅表到流转换的文档。

在CassandraAppendTableSink插入所有行至少一次到Cassandra的表，如果检查点已启用。但是，您可以将查询指定为upsert查询。

要使用CassandraAppendTableSink，必须将Cassandra连接器依赖项（flink-connector-cassandra）添加到项目中。以下示例显示了如何使用CassandraAppendTableSink。

Java
Scala

ClusterBuilder builder = ... // configure Cassandra cluster connection
CassandraAppendTableSink sink = new CassandraAppendTableSink(
  builder,
  // the query must match the schema of the table
  INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?));
Table table = ...
table.writeToSink(sink);

val builder: ClusterBuilder = ... // configure Cassandra cluster connection 
val sink: CassandraAppendTableSink = new CassandraAppendTableSink(
  builder,
  // the query must match the schema of the table
  INSERT INTO flink.myTable (id, name, value) VALUES (?, ?, ?))
val table: Table = ???
table.writeToSink(sink)