# 查询和写入Paimon表数据
#### 注意事项
- 需要底层文件系统（如OBS）可用。
- 主键表必须定义PRIMARY KEY，且主键列不能为NULL。
- 流式写入Paimon时需要开启checkpoint，数据在checkpoint时提交。
- 主键表的分桶数（bucket）建议与Sink并行度一致，避免数据倾斜。
- Paimon的分区键必须包含在主键中。
- Flink Streaming作业不支持insert overwrite。
- Paimon表使用Catalog方式管理元数据后，不允许rename。
- Paimon不支持Sink materializer，请配置table.exec.sink.upsert-materialize=NONE。
 
#### 代码示例
- **示例1：无主键数据追加写入Paimon Append表。**
  ```
  CREATE CATALOG lf_paimon_catalog
  WITH
    (
      'type' = 'paimon',
      'metastore' = 'lakeformation'
    );
  CREATE DATABASE IF NOT EXISTS lf_paimon_catalog.paimon_db
  WITH (
    'location' = 'obs://bucket/path/paimon-warehouse/paimon_db'
  );
  CREATE TABLE IF NOT EXISTS lf_paimon_catalog.paimon_db.paimon_table (
    id BIGINT,
    customer_id STRING,
    amount DOUBLE,
    region STRING
  );
  CREATE TABLE kafka_source(
    id BIGINT,
    customer_id STRING,
    amount DOUBLE,
    region STRING
  )
  WITH
    (
      'connector' = 'kafka',
      'topic' = 'user-events',
      'properties.bootstrap.servers' = '127.0.0.1:9092',
      'properties.group.id' = 'flink-consumer-group',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json'
    );
  INSERT INTO
    lf_paimon_catalog.paimon_db.paimon_table 
  SELECT
    *
  FROM
    kafka_source;
  ```
  
 
- **示例2：读取Paimon Append表。**
  ```
  CREATE CATALOG lf_paimon_catalog
  WITH
    (
      'type' = 'paimon',
      'metastore' = 'lakeformation'
    );
  CREATE TABLE sink (
    id BIGINT,
    customer_id STRING,
    amount DOUBLE,
    region STRING
  )
  WITH
    ('connector' = 'print');
  INSERT INTO
    sink
  SELECT
    *
  FROM
    lf_paimon_catalog.paimon_db.paimon_table /*+ OPTIONS('streaming'='true', 'monitor-interval'='60s') */;
  ```