### 1. 建表
1. text格式
    ```sql
    CREATE external table ods_crawler_google_keyword_bybatch_outtable_di (
        msg string
    )
    PARTITIONED BY 
    (ymd string,batch string)
    row format serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
    stored AS textfile
    location 'oss://oss-cn-shanghai-internal.aliyuncs.com/xxx/';
    ;
    ```
1. json格式

    ```sql
    CREATE EXTERNAL TABLE IF NOT EXISTS dialogue.dwd_ext_xingye_log_chatml_data(
        `data` STRING
    ) 
    PARTITIONED BY (
        data_type STRING,
        ymd STRING) 
    ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' 
    STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat' 
    OUTPUTFORMAT 'org.apache.hadoop.mapred.TextOutputFormat' 
    LOCATION 'oss://data-dlc/minimax-dialogue/data/chatml/';
    ```

1. parquet格式
   ```sql
   CREATE EXTERNAL TABLE dwd_crawler_html_out_di
   (
      url string,
       html string,
       source_tag string,
       lang string
   )
   partitioned BY (ymd string)
   stored AS parquet  
   location 'oss://data-dlc/minimax-dialogue/data/chatml/data_type=html_data/';
   ```
1. 自定义解析器
    ```sql
    CREATE EXTERNAL TABLE ods_crawler_sites_warc_data_extract_result_external_hi (
        warc_host string,
        meta_info string,
        final_url string,
        status_code string,
        data_dt string,
        biz_code string,
        target_url string,
        warc_date string,
        content_type string,
        content_length string,
        warc_file_name string,
        content string,
        import_msg string
    )
    PARTITIONED BY (
        host string,
        status string,
        ymd string,
        hour string
    )
    stored BY 'com.crawler.udf.sites_warc.WarcStorageHandler' 
    with serdeproperties (
        'max_file_size' = '157286400'
    ) 
    LOCATION 'oss://crawler-data-storage/Crawlerlee/'
    USING 'crawler_udf-1.0-SNAPSHOT.jar'
    ```
### 2. 使用
1. 插入数据
    ```sql
    SET odps.sql.unstructured.oss.commit.mode = true; -- 防止生成.odps文件夹
    INSERT OVERWRITE TABLE dwd_crawler_html_out_di PARTITION(ymd='${P_DATE}')
    SELECT 
        a.source_url,
        html,
        biz_code,
        parse_lang
    FROM 
        dwd_crawler_content_filtered_detail_byscore_di a
    WHERE 
        a.ymd = '${P_DATE}'
        AND a.score_level != 'low'
        AND COALESCE(html, '') != ''
    ```
2. 读取数据
    ```
    -- 扫描所有路径
    msck TABLE ods_crawler_sites_warc_data_extract_result_external_hi ADD partitions;
    -- 扫描指定路径
    ALTER TABLE ods_crawler_sites_warc_data_extract_result_external_hi ADD IF NOT EXISTS 
    PARTITION (host='demo.com',status='error',ymd='20250624', hour='09') 
    PARTITION (host='demo.com',status='success',ymd='20250624', hour='09')
    ;
    -- 指定location
    ALTER TABLE log_table_external ADD PARTITION (year = '2016', month = '06', day = '01')
    location 'oss://oss-cn-hangzhou-internal.aliyuncs.com/bucket名称/oss-odps-test/log_data_customized/2016/06/01/';
    -- 指定路径映射
    MSCK REPAIR TABLE  ods_crawler_sites_warc_data_extract_result_external_hi ADD PARTITIONS
    WITH PROPERTIES ('odps.msck.partition.column.mapping'='host:host,status:status,ymd:ymd');
    ```
3. load和unload
- load
```sql
create table xxx lifecycle 10 AS 
SELECT 
    '' AS msg
;
load overwrite table xxx
from
location 'oss://xxx/'
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'

```
- unload
```sql
SET odps.sql.unstructured.oss.commit.mode = true;
--导出数据。
UNLOAD FROM  
(
    SELECT * FROM xxx
)
INTO
LOCATION 'oss://xxx/'
stored AS TEXTFILE
;
```
5. 大字段导出
  - 建表增加参数
  ```sql
  tblproperties( 
      'mcfed.parquet.block.row.count.limit' = '100', 
      'mcfed.parquet.page.size.row.check.max' = '100', 
      'mcfed.parquet.page.size.row.check.min' = '100'
  )
  ```
        
  - 导出时增加参数
  ```sql
set odps.isolation.session.enable=true;
set odps.stage.mapper.mem=8192;
set odps.stage.mapper.jvm.mem=6144;
  ```
 5. 复杂parquet字段导入
 - 使用Schemaless Query读取数据 通过to_json转字符串
 ```sql
 INSERT OVERWRITE TABLE tmp_data_0319_100
SELECT * FROM 
LOCATION 'oss://oss-cn-shanghai-internal.aliyuncs.com/xxx/' 
(
  'file_format'='parquet'
)
;
SELECT 
    to_json(data)
FROM 
    tmp_data_0319_100
 ```
6. 设置导出文件数
`set odps.sql.split.dop = {"*":10};`

MaxCompute外表使用

目前Dataworks只支持写入doris，不支持读
从阿里云selectdb文档可以看到有一种导出方式：[SELECT INTO OUTFILE](https://help.aliyun.com/zh/selectdb/select-into-outfile?spm=a2c4g.11186623.help-menu-2503500.d_2_1_2_0.9cd416c0YJf33d)
可以搭配Dataworks的数据库节点：Doris节点+oss外表的方式导入数据到MaxCompute
## 1. 新建doris节点

可以先写个简单sql测试下是否可以查询到数据
参考sql
```sql
SET sql_select_limit = 999999999; -- sql结果条数限制
set enable_parallel_outfile = true; -- 开发并发，实际有无并发可参考官方文档查看
SET query_timeout = 900*3; -- 设置查询超时时间

SELECT 
    *
FROM  
    mm_hailuo.trace_log  
WHERE 
    create_date = '${P_DATE}' 
    AND GET_JSON_BIGINT(content, '$.type') = 3
INTO OUTFILE "s3://crawler-data-storage/trace_log/ymd=${P_DATE}/"
FORMAT AS PARQUET
PROPERTIES
(
    "s3.endpoint" = "oss-cn-shanghai-internal.aliyuncs.com",
    "s3.access_key" = "xxx",
    "s3.secret_key" = "xxx",
    "s3.region" = "cn-shanghai",
    "success_file_name" = "_SUCCESS"
);
```
## 2. 外表导入
[Schemaless Query](https://help.aliyun.com/zh/maxcompute/user-guide/schemaless-query?spm=a2c4g.11186623.help-menu-27797.d_2_1_4_1_11.733d3df4DrxUWU&scm=20140722.H_2927126._.OR_help-T_cn~zh-V_1)
MaxCompute支持免定义查询（Schemaless Query）OSS中Parquet外部表数据，解析后的数据集结果可以导出至OSS或写入内部表，同时也可以作为子查询嵌入SQL运算，从而便捷地操作湖上数据。
```sql
INSERT OVERWRITE TABLE ods_hailuo_trace_log_di PARTITION(ymd='${P_DATE}')
SELECT 
    create_date,
    biz_id,
    log_key,
    sub_log_key,
    content
FROM location 'oss://oss-cn-shanghai-internal.aliyuncs.com/crawler-data-storage/trace_log/ymd=${P_DATE}/'
(
'file_format'='parquet',
'file_pattern_blacklist'='.*SUCCESS'
);
```

Doris数据同步到MaxCompute

### 1. range分区
1. 建表
```sql
CREATE TABLE `xxx` (
    `ymd` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
    user_id varchar(100) COMMENT '用户ID',
    platform varchar(50) COMMENT '平台',
    is_new varchar(20) COMMENT '是否新用户',
    sub_tab_name varchar(128) COMMENT '发现页tab',
    page_view_cnt BIGINT COMMENT '发现页曝光PV',
    search_topic_npc_view_cnt BIGINT COMMENT '发现页NPC曝光PV',
    search_topic_npc_view_npc_cnt BIGINT COMMENT '发现页NPC曝光npc数',
    search_topic_npc_click_cnt BIGINT COMMENT '发现页NPC点击PV', 
    search_topic_npc_click_npc_cnt BIGINT COMMENT '发现页NPC点击npc数',
    message_send_npc_cnt BIGINT COMMENT '消息发送npc次数',
    message_send_message_cnt BIGINT COMMENT '消息发送消息次数'
) ENGINE=OLAP
DUPLICATE KEY(`ymd`,`user_id`,`platform`,`is_new`,`sub_tab_name`)
PARTITION BY RANGE(`ymd`)()
DISTRIBUTED BY HASH(`user_id`) BUCKETS 10
PROPERTIES (
    "replication_allocation" = "tag.location.default: 3", -- 副本分配策略，在标签为default的位置保存3个副本
    "min_load_replica_num" = "-1", -- 最小导入副本数，-1表示使用默认值（通常等于副本数）
    "dynamic_partition.enable" = "true", -- 启用动态分区功能
    "dynamic_partition.time_unit" = "DAY", -- 分区时间单位为天（DAY）
    "dynamic_partition.time_zone" = "Asia/Shanghai", -- 时区设置为上海时区（UTC+8）
    "dynamic_partition.start" = "-15", -- 保留当前日期前15天的分区
    "dynamic_partition.end" = "7", -- 提前创建未来7天的分区
    "dynamic_partition.prefix" = "p_", -- 分区名前缀，生成的分区名如p_20250126
    "dynamic_partition.replication_allocation" = "tag.location.default: 3", -- 动态创建的分区也保持3副本
    "dynamic_partition.buckets" = "10", -- 每个动态分区的bucket数量为10
    "dynamic_partition.create_history_partition" = "false", -- 不创建历史分区（只创建当前和未来分区）
    "dynamic_partition.history_partition_num" = "-1", --  -1表示不限制历史分区数量
    "dynamic_partition.hot_partition_num" = "0", -- 0表示没有特别的热分区设置
    "dynamic_partition.reserved_history_periods" = "NULL", -- 保留的历史周期，NULL表示使用默认设置
    "dynamic_partition.storage_policy" = "", -- 存储策略为空（使用默认）
    "storage_medium" = "hdd", -- 存储介质为机械硬盘（hdd）
    "storage_format" = "V2", -- 使用V2存储格式
    "light_schema_change" = "true", -- 启用轻量级schema变更
    "disable_auto_compaction" = "false", -- 不禁用自动压缩
    "enable_single_replica_compaction" = "false", -- 不启用单副本压缩
    "group_commit_interval_ms" = "10000", -- 组提交时间间隔10秒
    "group_commit_data_bytes" = "134217728" -- 组提交数据大小阈值128MB
);
```
2. 临时开启历史分区创建
```sql
ALTER TABLE xxx SET (
    "dynamic_partition.start" = "-90",
    "dynamic_partition.create_history_partition" = "true"
);
-- 等待一段时间 （10分钟左右）
SHOW PARTITIONS FROM xxx;

ALTER TABLE xxx SET (
    "dynamic_partition.create_history_partition" = "false"
);
```
### 1. list分区
```sql
CREATE TABLE `xxx` (
    ymd varchar(50) NULL,
    user_id varchar(100) COMMENT '用户ID',
    platform varchar(50) COMMENT '平台',
    is_new varchar(20) COMMENT '是否新用户',
    sub_tab_name varchar(128) COMMENT '发现页tab',
    page_view_cnt BIGINT COMMENT '发现页曝光PV',
    search_topic_npc_view_cnt BIGINT COMMENT '发现页NPC曝光PV',
    search_topic_npc_view_npc_cnt BIGINT COMMENT '发现页NPC曝光npc数',
    search_topic_npc_click_cnt BIGINT COMMENT '发现页NPC点击PV', 
    search_topic_npc_click_npc_cnt BIGINT COMMENT '发现页NPC点击npc数',
    message_send_npc_cnt BIGINT COMMENT '消息发送npc次数',
    message_send_message_cnt BIGINT COMMENT '消息发送消息次数'

) ENGINE=OLAP
DUPLICATE KEY(`ymd`,`user_id`,`platform`,`is_new`,`sub_tab_name`)
PARTITION BY LIST (`ymd`)
(
  PARTITION p20250626 VALUES IN ('2025-06-26')
)
PROPERTIES (
    "file_cache_ttl_seconds" = "0",
    "is_being_synced" = "false",
    "storage_medium" = "hdd",
    "storage_format" = "V2",
    "inverted_index_storage_format" = "V2",
    "light_schema_change" = "true",
    "disable_auto_compaction" = "false",
    "enable_single_replica_compaction" = "false",
    "group_commit_interval_ms" = "10000",
    "group_commit_data_bytes" = "134217728"
);
delete from  xxx where ymd = "${P_DATE}" ;
alter table xxx add partition p_${p_date} VALUES IN (("${P_DATE}"))
```


Doris分区使用

```sql

WITH hash_bucket AS (
    SELECT  
        *,
        ROW_NUMBER() OVER (PARTITION BY bucket_no) AS bucket_rel_index,
        COUNT(1) OVER (PARTITION BY bucket_no ) AS bucket_size
    FROM    
    (
        SELECT  
            *,
            ABS(HASH(cid)) % 100000 AS bucket_no 
        FROM    
            xxxx        
    ) 
),
bucket_base AS 
(
    SELECT
        bucket_no,
        SUM(bucket_size) OVER (ORDER BY bucket_no ASC) - bucket_size AS bucket_base
    FROM
    (
        SELECT 
            DISTINCT bucket_no,
            bucket_size
        FROM
            hash_bucket
    )
)
SELECT
    t1.*,
    t2.bucket_base + bucket_rel_index AS id_index
FROM
    hash_bucket t1 
JOIN 
    bucket_base t2 
    ON t1.bucket_no = t2.bucket_no
;
```

核心代码如上，主要思路先做分桶进行分布式局部排序、再基于桶大小得到全局索引

1. 第一步先对id做hash分桶，然后计算每个桶的大小，桶数量可以根据需要计算的id数量来评估，这里分100000个桶，然后计算出每个id在桶内的相对位置bucket_rel_index，同时计算出桶大小bucket_size；
2. 第二步根据桶大小计算每个桶的基址；
3. 第三步将桶基址+id在桶内的相对地址得到全局唯一的绝对地址id_index；



全局排序优化

优化前：
```sql
SELECT 
        /*+ mapjoin(dd),skewjoin(a(url))*/
        a.* except(url_tag, join_key),
        CASE 
            WHEN cc.url IS NOT NULL THEN 'black_url'
            WHEN dd.url IS NOT NULL THEN 'black_url'
            ELSE ''
        END AS url_tag
    FROM
    (
        select 
            *,
            1 AS join_key
        from 
            dwd_crawler_urls_step1_di a 
        where 
            ymd = '${P_DATE}'
            AND url_tag != 'black_url'
    ) as
    LEFT JOIN 
        black_urls2 cc  
        ON replace(a.host, 'www.', '') = replace(cc.url, 'www.', '')
	LEFT JOIN 
        black_urls1 dd  
        ON INSTR(a.url, dd.url) > 0
```
优化后
```sql
SELECT 
        /*+ mapjoin(dd),skewjoin(a(url))*/
        a.* except(url_tag, join_key),
        CASE 
            WHEN cc.url IS NOT NULL THEN 'black_url'
            WHEN dd.url IS NOT NULL THEN 'black_url'
            ELSE ''
        END AS url_tag
    FROM
    (
        select 
            *,
            1 AS join_key
        from 
            dwd_crawler_urls_step1_di a 
        where 
            ymd = '${P_DATE}'
            AND url_tag != 'black_url'
    ) a
    LEFT JOIN 
        black_urls1 dd  
        ON INSTR(a.url, dd.url) > 0
    LEFT JOIN 
        black_urls2 cc  
        ON replace(a.host, 'www.', '') = replace(cc.url, 'www.', '')
```
优化前执行时间

<img width="991" alt="Image" src="https://github.com/user-attachments/assets/3d62cdfd-b1e0-4579-adc9-a8bc19ac64ff" />

<img width="872" alt="Image" src="https://github.com/user-attachments/assets/46033d2f-9fbe-4400-b6c6-489f97ee7f00" />

优化后执行时间

<img width="978" alt="Image" src="https://github.com/user-attachments/assets/74de5fc3-08ff-4ed6-8c94-9b0d31c18799" />

<img width="944" alt="Image" src="https://github.com/user-attachments/assets/b2f397da-420d-484b-9655-3dda6ee57046" />
