#### 1. 取最近一个不为空的值填充
```sql
WITH mian_data AS (
    SELECT
        c0,
        c1,
        IF(c1 = '处置', NULL, c2) AS c2
    FROM
        VALUES
         (1,'处置','1')
        ,(2,'处置','2')
        ,(3,'处置','3')
        ,(4,'处置','4')
        ,(5,'提交','A')
        ,(6,'处置','5')
        ,(7,'提交','b')
    AS a(c0,c1,c2)
)
SELECT 
    *,
    FIRST_VALUE(c2, True) over(order by c0 ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING)
FROM 
    mian_data 
```

窗口函数的一些使用


### 1. 数仓分层
按 ods -> dwd -> dwm -> dws ->ads 来做
- dwd：事实级别的明细，或者did uid级的数据。比如某个功能的清洗后标准化的明细。比如日活表 新增表
- dwm：聚合后的数据，我理解所有从did uid聚合到没有uid did的，都可以放这层里
- dws：各种宽表，这里就不限制粒度了吧，聚合后的指标宽表，或者did uid粒度的大宽表都放在这层里
- ads：看板数据
### 2. 业务侧看板命名
- dwd表，在_di前带上detail关键字，代表是事实明细
- dwm表，带上stats的关键字，代表是聚合后的一些指标
- ads表，带上metric关键字，如果是一些核心功能的看板，会带上core_metric。

表命名规范

# 阿里云个人使用的一些姿势

## 一、登录

[登录 dataworks](https://dataworks.console.aliyun.com/workspace/list)，选择对应工作空间进入，显示无空间找管理员开通

<img width="1693" alt="Image" src="https://github.com/user-attachments/assets/b2e470eb-06e2-4c2c-8d12-da6232adc7b2" />

<img width="1238" alt="Image" src="https://github.com/user-attachments/assets/2408c27a-fcf3-4631-a68a-2f3005c49936" />

## 二、临时 Sql 运行

临时查询下新建一个目录，在下面新建 odps sql 节点即可

## 三、周期任务

### sql 任务

业务流程 MaxCompute 数据开发目录下新建 ODPS Sql 节点

### PySpark 简单模式

[官方文档](https://help.aliyun.com/zh/maxcompute/user-guide/develop-a-spark-on-maxcompute-application-by-using-pyspark?spm=a2c4g.11186623.0.i2)

1. 资源中新建 python 脚本
 
<img width="443" alt="Image" src="https://github.com/user-attachments/assets/9e325886-5153-4c75-b19b-23fa6476f1b6" />

 ```python
from pyspark.sql import SparkSession
if __name__ == '__main__':
spark = SparkSession.builder.appName("spark sql").getOrCreate()
spark.sql("SELECT 1").show()
```
2. 数据开发文件夹新建odps spark节点

<img width="735" alt="Image" src="https://github.com/user-attachments/assets/e4a30db5-a9b3-4110-bd83-85c764450f39" />

发布上线皆可，sparkui会在日志中打印

### PySpark兼容模式

阿里云的PySpark是基于Python 3.7 + Spark2的，而且看上去已经疏于维护了。新任务建议直接用PyODPS3，而历史包袱较重的旧任务，如需从AWS快速迁移，则可以参考如下方式：

1. 资源中新建python脚本（和简单模式相同）

2. 数据开发文件夹新建odps spark节点（和简单模式相同），语言选择Python，spark版本选择Spark 3.x，然后添加Python脚本作为主Python资源

3. 点击【配置项】的【添加一条】按钮，依次添加如下配置（全都是必须添加的配置）：

<table>
<tr>
<td>配置项 </td><td>配置值 </td></tr>
<tr>
<td>spark.hadoop.odps.cupid.resources </td><td>talkie_ug.python310-spark3.tar.gz </td></tr>
<tr>
<td>spark.pyspark.python </td><td>./talkie_ug.python310-spark3.tar.gz/python3/bin/python3.10 </td></tr>
<tr>
<td>spark.executorEnv.LD_LIBRARY_PATH </td><td>$LD_LIBRARY_PATH:./talkie_ug.python310-spark3.tar.gz/python3/3rdlib </td></tr>
<tr>
<td>spark.yarn.appMasterEnv.LD_LIBRARY_PATH </td><td>$LD_LIBRARY_PATH:./talkie_ug.python310-spark3.tar.gz/python3/3rdlib </td></tr>
</table>

添加完成后，配置其他业务参数（如p_date），即可运行。

![Image](https://github.com/user-attachments/assets/eccb712b-583a-4123-b1d7-23b23f699c63)

### Python pyodps节点

其他都差不多，主要是参数和第三方包

参数：args['p_date'] 

第三方包：可以在调度资源组上安装

写表：
```python
write_records=[] #二维数组
if len(write_records)>0:
 _table = o.get_table('dw_risk_warehouse.ods_meta_cwf_node_info_df')
 _table.delete_partition('pt={0}'.format(pt), if_exists=True)
 with _table.open_writer(partition='pt={0}'.format(pt), create_partition=True) as writer:
 writer.write(write_records)
 print('插入数据成功')
```
读表：
```python
pt='20221025'
query_sql = '''
 SELECT *
 FROM xxx
 WHERE pt = '{0}'
 '''.format(pt)
instance = o.run_sql(query_sql, hints={'odps.instance.priority': 0})
logview = instance.get_logview_address()
print(logview)
instance.wait_for_success()
with instance.open_reader(tunnel=True) as reader:
 for index,record in enumerate(reader):
 try:
 node_id = record[0]
 database_name = record[1]
 node_name = record[2]
 except:
 pass
```

### Udf(python)
1. 进入数据开发界面，在业务流程下的 maxcompute/资源 下新建资源，类型选择python
	
<img width="489" alt="Image" src="https://github.com/user-attachments/assets/82b13fc9-86a8-4c8a-b2c8-62de3b963ed1" />

<img width="677" alt="Image" src="https://github.com/user-attachments/assets/f6e757dd-eca7-4e05-8a16-fc507ab0728f" />

2. 编写代码，[官方文档](https://help.aliyun.com/zh/maxcompute/user-guide/python-3-udfs?spm=a2c4g.11174283.0.i2)，简单示例如下
```python
from odps.udf import annotate
@annotate("string->bigint")
class GetTextLen(object):
 def evaluate(self, input_str):
 try:
 if not input_str or input_str == '' or input_str is None:
 return 0
 return len(input_str.split())
 except:
 return 0
```

写好之后保存并提交

1. 在函数下新建函数，内容如图

<img width="478" alt="Image" src="https://github.com/user-attachments/assets/43fa3b37-4a01-42e0-8cb2-d12ed90a9314" />

<img width="922" alt="Image" src="https://github.com/user-attachments/assets/d31dbbb0-e506-4267-9857-0ba3194b3a0d" />

2. 使用
 python2 版本
```sql
 SELECT talkie_dialogue.GET_TEXT_LEN('34d dfds'); -- 跨空间使用需要加库名，本空间不需要
```
 python3 版本，这两句选中一起跑
```sql
set odps.sql.python.version=cp37; -- cp311
SELECT talkie_dialogue.GET_TEXT_LEN('34d dfds')
```

## 四、运维中心

点击数据开发界面的运维中心，或者在已上线的任务右键运维中心

主要用下红框的几个模块

<img width="1873" alt="Image" src="https://github.com/user-attachments/assets/2b62f220-7e48-4852-8881-80827b1f4d3b" />

1. 运维中心可以看到任务概览
2. 周期任务：即在数据开发界面提交的任务，可进行下线、修改责任人、冻结等操作，也可以补数据
3. 周期实例：阿里云根据周期任务的调度配置生成的实例，可进行实例状态查看、实例冻结、重跑、置成功等操作
4. 补数据：查看补数据任务情况和手动补数据
5. 规则管理：可设置任务出错、超时、未开始等监控

## 五、数据地图

全部产品中选中数据地图进入

或者[链接](https://dmc-us-east-1.data.aliyun.com/dm)

<img width="1293" alt="Image" src="https://github.com/user-attachments/assets/31523a55-57cb-489d-a084-b61657ffe7fb" />


搜索界面可以查看库中有哪些表

<img width="1898" alt="Image" src="https://github.com/user-attachments/assets/c1e98372-356e-4a63-a9b4-55c8b5a0eab9" />

可以看到表相关信息

<img width="1860" alt="Image" src="https://github.com/user-attachments/assets/5a1b2940-746d-49f3-ae40-f2b9617e88e4" />

1. 产出任务：有对应空间的访客权限即可看到任务脚本，在需要依赖的时候，可以复制括号中的节点 id 进行依赖
2. 字段信息中的生成 ddl，可以生成建表语句
3. 分区信息：可以看到分区的相关信息
4. 血缘关系：可供参考，可能不准
5. 数据质量：配置了 DQC 之后可以看到每天检查的数据质量情况
6. 数据预览
7. 权限申请：申请表级或者字段级的数据权限，查看申请 select 和 describe 即可

<img width="1482" alt="Image" src="https://github.com/user-attachments/assets/97f42218-0413-49c6-b9a5-5f67ec30a18a" />

<img width="1516" alt="Image" src="https://github.com/user-attachments/assets/b507fff9-bdb5-4520-acce-3c070266f0a0" />

## 六、空间设置

每人独立的设置

主要可以调整下调度设置

<img width="550" alt="Image" src="https://github.com/user-attachments/assets/5a1fa25d-daf4-4f78-a6d7-45c3d424e6d8" />

<img width="911" alt="Image" src="https://github.com/user-attachments/assets/8302d29a-223a-4080-8445-4582df2f43aa" />

## 七、脚本变动

### FROM_UNIXTIME

```python
FROM_UNIXTIME(create_time / 1000, 'yyyy-MM-dd')
改为
TO_CHAR(FROM_UNIXTIME(BIGINT(create_time / 1000)),'yyyy-mm-dd')
并且这里FROM_UNIXTIME只接受BIGINT类型
```

### GET_VERSION

已经在 talkie_da 库下建好，应该直接使用 talkie_da.GET_VERSION(app_version) 即可计算

### DATEDIFF

```python
DATEDIFF('${P_DATE}', a.min_connect_date)
改为
DATEDIFF(TO_DATE('${P_DATE}','yyyy-mm-dd'), TO_DATE(a.min_connect_date,'yyyy-mm-dd'), 'dd')
```

>

### INSERT OVERWRITE TABLE

INSERT OVERWRITE 表名 改成 INSERT OVERWRITE TABLE 表名

### LATERAL VIEW EXPLODE 语句，exploded_table 别名不能省略

Spark sql,exploded_table 可以不写，但是阿里云不行

```sql
SELECT key_column, value_column
FROM your_table
LATERAL VIEW EXPLODE(map_column) exploded_table AS key_column, value_column;
```

### APPROX_PERCENTILE

改为 [PERCENTILE_APPROX](https://help.aliyun.com/zh/maxcompute/user-guide/aggregate-functions?spm=a2c4g.11186623.0.i3#section-do0-b0t-s3q)

### 数据类型转换

阿里云表定义数据类型和插入数据类型不一致时候会报错，int 和 bigint 也算作不同的类型。bigint 到 double 也不会做自动转换

<img width="1315" alt="Image" src="https://github.com/user-attachments/assets/4c79e957-d621-4eaa-9b4a-424a54b2b668" />

## 八、AWS 历史数据同步到 MC

### 在 mc 上建外表

1. 请表名前面加上 glue_前缀用来区分
2. oss(阿里云的对象存储)路径：
 整体结构和 s3 一致
3. 数据目前应该是同步到了 05-26，大家先用，我看下能否对关键表搞成例行同步的，这样就能随时读到最新的分区了。

```sql
create external table glue_<aws_table_name>
(
 -- 这里字段和aws建表语句保持一致即可
 xxx string,
 yyy bigint,
 ...
)
PARTITIONED BY ( 
 ymd string COMMENT ''
)
stored as parquet 
location 'oss://minimax-data-analysis/db/<aws_db_name>/<aws_table_name>';
```

### 修复分区

```sql
msck repair table glue_<aws_table_name> add partitions
```

注意看执行日志，会有 add xxx partitions 的提示，出现这行日志的时候就代表修复成功了

## 九、阿里云导出到 clickhouse

示例任务：ads_talkie_launch_metric_di 同步

![Image](https://github.com/user-attachments/assets/bed23c44-15f5-492e-8467-745d70011e79)

十、python udf 使用第三方包

> [https://help.aliyun.com/zh/maxcompute/user-guide/reference-a-third-party-package-in-a-pyodps-node-1](https://help.aliyun.com/zh/maxcompute/user-guide/reference-a-third-party-package-in-a-pyodps-node-1)

使用 docker 打包工具

`pyodps-pack -o 包名.tar.gz 包名`

udf 内容：

```python
import sys
from odps.udf import annotate

@annotate("double->double")class MyPsi(object):
 def __init__(self):
 _# 将路径增加到引用路径_
 sys.path.insert(0, "work/包名.tar.gz/packages")

 def evaluate(self, arg0):
 _# 将 import 语句保持在 evaluate 函数内部_from scipy.special import psi

 return float(psi(arg0))
```