# Orc Format
#### 功能描述
Apache Orc Format允许读写ORC数据。更多具体使用可参考开源社区文档：[Orc Format](https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/table/formats/orc/)。
#### 支持的Connector
- FileSystem
 
#### 参数说明
表1参数说明 
| 参数     | 是否必选 | 默认值 | 类型     | 描述                    |
|:---|:---|:---|:---|:---|
| format | 是    | 无   | String | 指定要使用的格式，这里应该是 'orc'。 |
   
Orc 格式也支持来源于 [Table properties](https://orc.apache.org/docs/hive-config.html#table-properties) 的表属性。 举个例子，您可以设置 orc.compress=SNAPPY 来允许spappy压缩。
#### 数据类型映射
Orc 格式类型的映射和 Apache Hive 是兼容的。下面的表格列出了 Flink 类型的数据和 Orc 类型的数据的映射关系。
表2数据类型映射 
| Flink数据类型 | Orc物理类型   | Orc逻辑类型   |
|:---|:---|:---|
| CHAR      | bytes     | CHAR      |
| VARCHAR   | bytes     | VARCHAR   |
| STRING    | bytes     | STRING    |
| BOOLEAN   | long      | BOOLEAN   |
| BYTES     | bytes     | BINARY    |
| DECIMAL   | decimal   | DECIMAL   |
| TINYINT   | long      | BYTE      |
| SMALLINT  | long      | SHORT     |
| INT       | long      | INT       |
| BIGINT    | long      | LONG      |
| FLOAT     | double    | FLOAT     |
| DOUBLE    | double    | DOUBLE    |
| DATE      | long      | DATE      |
| TIMESTAMP | timestamp | TIMESTAMP |
| ARRAY     | -         | LIST      |
| MAP       | -         | MAP       |
| ROW       | -         | STRUCT    |
   
#### 示例
使用kafka发送数据，输出到print中。
1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列 \> 找到作业的所属队列 \> 更多 \> 测试地址连通性 \> 输入kafka的地址 \> 测试）。如果能连通，则表示跨源已经绑定成功；否则表示未成功。
2. 创建flink opensource sql作业，**开启checkpoint** ，并提交运行，其代码如下：
   
   ```
   CREATE TABLE kafkaSource (
     order_id string,
     order_channel string,
     order_time string, 
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'kafka',
     'topic-pattern' = kafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId'',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'csv'
   );
   CREATE TABLE sink (
     order_id string,
     order_channel string,
     order_time string, 
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'filesystem',
     'format' = 'orc',
     'path' = 'obs://xx'
   );
   insert into sink select * from kafkaSource;
   ```
   
   
3. 向kafka的作为source的topic中插入下列数据： 
   ```
   202103251505050001,appshop,2021-03-25 15:05:05,500.00,400.00,2021-03-25 15:10:00,0003,Cindy,330108
   202103241606060001,appShop,2021-03-24 16:06:06,200.00,180.00,2021-03-24 16:10:06,0001,Alice,330106
   ```
   
   
4. 读取sink表中配置的obs路径中的orc文件，其数据结果如下 
   ```
   202103251202020001, miniAppShop, 2021-03-25 12:02:02, 60.0, 60.0, 2021-03-25 12:03:00, 0002, Bob, 330110
   202103241606060001, appShop, 2021-03-24 16:06:06, 200.0, 180.0, 2021-03-24 16:10:06, 0001, Alice, 330106
   ```