1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254 | from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import timedelta,datetime
import pendulum
import pytz
tz = pytz.timezone('Asia/Shanghai')
default_args = {
'owner': 'Terry',
'depends_on_past': False,
'start_date': pendulum.datetime(year=2024, month=6, day=17).astimezone(tz),
'email': ['tian23.li@tcl.com'],
'email_on_failure': True,
'email_on_retry': True,
'retries': 2,
'retry_delay': timedelta(minutes=20),
}
_config = {
'conf': {
'spark.executor.extraJavaOptions': '-XX:+UseConcMarkSweepGC',
"spark.dynamicAllocation.minExecutors":8,
"spark.dynamicAllocation.maxExecutors":8,
"spark.yarn.executor.memoryOverhead":2648,
"spark.driver.maxResultSize": "4g",
"spark.yarn.maxAppAttempts": 1,
"spark.network.timeout": 1000000,
"spark.executor.heartbeatInterval" : 1000000
},
'executor_cores': 4,
'executor_memory': '6g',
'name': '{{ task_instance.task_id }}',
'driver_memory': '6g',
}
dag = DAG(
dag_id='multi_interest_youku_jx_lst',
description='flow for multi-interest youku jx lst',
default_args=default_args,
schedule_interval='00 06 * * *',
)
# kill_iPython_job = BashOperator(
# task_id='kill_iPython_job',
# bash_command="ps -ef | grep ipykernel | grep -v grep | awk '{print $2}' | xargs kill -9",
# #bash_command="kill $(nvidia-smi -g 0)",
# dag=dag,
# )
get_youku_vip_users = BashOperator(
task_id='get_youku_vip_users',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 0_get_youku_vip_users.py",
depends_on_past =False,
dag=dag,
)
update_youku_jx_lst_media = BashOperator(
task_id='update_youku_jx_lst_media',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/recsys2/bin/python 0_update_youku_jx_lst_media.py",
depends_on_past =False,
dag=dag,
)
update_content_embedding_youku = BashOperator(
task_id='update_content_embedding_youku',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/script && /data/dev/miniconda/envs/easyrec/bin/python update_content_embedding_youku.py",
depends_on_past =False,
dag=dag,
)
update_content_embedding_youku_nomic = BashOperator(
task_id='update_content_embedding_youku_nomic',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/script && /data/dev/miniconda/envs/easyrec/bin/python update_content_embedding_youku_nomic.py",
depends_on_past =False,
dag=dag,
)
save_youku_nomic_to_hive = SparkSubmitOperator(
task_id='save_youku_nomic_to_hive',
dag=dag,
**_config,
application= '/data/gangyanyuan/terry/terry-recsys-2024/script/save_youku_nomic_to_hive.py',
conn_id = "spark_local",
)
get_ecpm_aids = BashOperator(
task_id='get_ecpm_aids',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/recsys2/bin/python get_ecpm_aids.py",
depends_on_past =False,
dag=dag,
)
generate_user_click_data_youku = BashOperator(
task_id='generate_user_click_data_youku',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/recsys2/bin/python 1_get_user_click_data_youku.py",
depends_on_past =False,
dag=dag,
)
# data reconstruction. for 18
update_user_click_youku_jx_lst_60days = BashOperator(
task_id='update_user_click_youku_jx_lst_60days',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 2_1_update_user_click_youku_jx_lst_60days.py",
depends_on_past =False,
dag=dag,
)
generate_diffurec_dataset_youku_jx_lst = BashOperator(
task_id='generate_diffurec_dataset_youku_jx_lst',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 3_generate_diffurec_dataset_youku_jx_lst.py",
depends_on_past =False,
dag=dag,
)
get_pretrained_content_embedding_youku = BashOperator(
task_id='get_pretrained_content_embedding_youku',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python get_pretrained_content_embedding_youku.py",
depends_on_past =False,
dag=dag,
)
pca_text = BashOperator(
task_id='pca_text',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script && /data/dev/miniconda/envs/deepmatch/bin/python pca.py --dataset youku_jx_lst --text_feat",
depends_on_past =False,
dag=dag,
)
# # LLM_Diffurec v2.0
train_diffurec_youku_jx_lst_novip = BashOperator(
task_id='train_diffurec_youku_jx_lst_novip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/DiffuRec/src_v2 && /data/dev/miniconda/envs/pt1.0/bin/python main_daily.py --dataset youku_jx_lst --max_len 10 --text_feat",
depends_on_past =False,
dag=dag,
)
generate_inference_dataset = BashOperator(
task_id='generate_inference_dataset',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 4_generate_inference_dataset.py",
depends_on_past =False,
dag=dag,
)
generate_llm_diffurec_recall_result_014578 = BashOperator(
task_id='generate_llm_diffurec_recall_result_014578',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/pt1.0/bin/python 5_generate_llm_diffurec_recall_result_014578.py --text_feat",
depends_on_past =False,
dag=dag,
)
# for 145036 78 & user_click_seq
update_user_click_youku_jx_lst_60days_013456 = BashOperator(
task_id='update_user_click_youku_jx_lst_60days_013456',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 2_1_update_user_click_youku_jx_lst_60days_013456.py",
depends_on_past =False,
dag=dag,
)
# generate_diffurec_dataset_youku_jx_lst_013456 = BashOperator(
# task_id='generate_diffurec_dataset_youku_jx_lst_013456',
# bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 3_generate_diffurec_dataset_youku_jx_lst_013456.py",
# depends_on_past =False,
# dag=dag,
# )
# # train
# train_diffurec_youku_jx_lst_013456 = BashOperator(
# task_id='train_diffurec_youku_jx_lst_013456',
# bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/DiffuRec/src && /data/dev/miniconda/envs/pt1.0/bin/python main_daily_youku_013456.py --dataset youku_jx_lst --max_len 10",
# depends_on_past =False,
# dag=dag,
# )
# generate_inference_dataset_013456 = BashOperator(
# task_id='generate_inference_dataset_013456',
# bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/deepmatch/bin/python 4_generate_inference_dataset_013456.py",
# depends_on_past =False,
# dag=dag,
# )
# generate_diffurec_recall_result_45 = BashOperator(
# task_id='generate_diffurec_recall_result_45',
# bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july && /data/dev/miniconda/envs/pt1.0/bin/python 5_generate_diffurec_recall_result_45.py",
# depends_on_past =False,
# dag=dag,
# )
save_youku_jx_lst_recall_to_hive = SparkSubmitOperator(
task_id='save_youku_jx_lst_recall_to_hive',
dag=dag,
**_config,
application= '/data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/save_youku_jx_lst_recall_to_hive.py',
conn_id = "spark_local",
)
# deploy vip model
get_pretrained_content_embedding_youku_vip = BashOperator(
task_id='get_pretrained_content_embedding_youku_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/vip && /data/dev/miniconda/envs/deepmatch/bin/python get_pretrained_content_embedding_youku_vip.py",
depends_on_past =False,
dag=dag,
)
get_pretrained_image_embedding_minicpm_youku_vip = BashOperator(
task_id='get_pretrained_image_embedding_minicpm_youku_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/vip && /data/dev/miniconda/envs/deepmatch/bin/python get_pretrained_image_embedding_minicpm_youku_vip.py",
depends_on_past =False,
dag=dag,
)
generate_diffurec_dataset_youku_jx_lst_vip = BashOperator(
task_id='generate_diffurec_dataset_youku_jx_lst_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/vip && /data/dev/miniconda/envs/deepmatch/bin/python 3_generate_diffurec_dataset_youku_jx_lst_vip.py",
depends_on_past =False,
dag=dag,
)
pca_text_image_vip = BashOperator(
task_id='pca_text_image_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script && /data/dev/miniconda/envs/deepmatch/bin/python pca_vip.py --dataset youku_jx_lst --text_feat --image_feat",
depends_on_past =False,
dag=dag,
)
train_diffurec_youku_jx_lst_vip = BashOperator(
task_id='train_diffurec_youku_jx_lst_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/DiffuRec/src_v2 && /data/dev/miniconda/envs/pt1.0/bin/python main_daily_vip.py --dataset youku_jx_lst --max_len 10 --text_feat --image_feat",
depends_on_past =False,
dag=dag,
)
generate_inference_dataset_vip = BashOperator(
task_id='generate_inference_dataset_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/vip && /data/dev/miniconda/envs/deepmatch/bin/python 4_generate_inference_dataset_vip.py",
depends_on_past =False,
dag=dag,
)
generate_diffurec_recall_result_vip = BashOperator(
task_id='generate_diffurec_recall_result_vip',
bash_command="cd /data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/vip && /data/dev/miniconda/envs/pt1.0/bin/python 5_generate_diffurec_recall_result_vip.py --text_feat --image_feat",
depends_on_past =False,
dag=dag,
)
save_user_click_youku_seq_to_hive = SparkSubmitOperator(
task_id='save_user_click_youku_seq_to_hive',
dag=dag,
**_config,
application= '/data/gangyanyuan/terry/terry-recsys-2024/multi-interest/script/youku_jx_july/save_user_click_youku_seq_to_hive.py',
conn_id = "spark_local",
)
get_youku_vip_users
update_youku_jx_lst_media >> update_content_embedding_youku_nomic >> save_youku_nomic_to_hive
update_youku_jx_lst_media >> update_content_embedding_youku >> get_ecpm_aids >> generate_user_click_data_youku >> update_user_click_youku_jx_lst_60days_013456 >> update_user_click_youku_jx_lst_60days >> generate_diffurec_dataset_youku_jx_lst >> get_pretrained_content_embedding_youku >> pca_text >> train_diffurec_youku_jx_lst_novip >> generate_inference_dataset >> generate_llm_diffurec_recall_result_014578
#generate_diffurec_dataset_youku_jx_lst_013456 >> train_diffurec_youku_jx_lst_013456 >> generate_inference_dataset_013456
update_user_click_youku_jx_lst_60days_013456 >> generate_diffurec_dataset_youku_jx_lst_vip >> get_pretrained_content_embedding_youku_vip >> get_pretrained_image_embedding_minicpm_youku_vip >> pca_text_image_vip >> train_diffurec_youku_jx_lst_vip >> generate_inference_dataset_vip >> generate_diffurec_recall_result_vip
[generate_llm_diffurec_recall_result_014578,generate_diffurec_recall_result_vip]>> save_youku_jx_lst_recall_to_hive >> save_user_click_youku_seq_to_hive
|