Plain Text
			[業務線]-[任務類型]-[環境]-[序號]示例：
• recmd-hive-prod-001（推薦業務線 Hive 生產任務）
• user-spark-prod-002（用户業務線 Spark 生產任務）
• log-flink-prod-003（日誌業務線 Flink 生產任務）
Task ID 需滿足：
• 全局唯一性
• 包含業務屬性，便於分類統計
• 與任務調度系統（如 Airflow、DolphinScheduler）中的任務 ID 保持一致
二、任務標識在計算引擎中的傳遞
2.1 Hive 任務
通過hiveconf傳遞任務標識，並配置 Hive 在訪問數據時攜帶該標識：
bash
Bash
			# 提交Hive SQL任務時指定Task ID
			beeline -u "jdbc:hive2://prod-hive:10000/prod_db;principal=hive/prod-hive@BIGDATA.COM" \--hiveconf task.id=recmd-hive-prod-001 \--hiveconf ranger.audit.taskid=recmd-hive-prod-001 \-f在 Hive 配置中關聯任務標識：
xml
XML
			<!-- hive-site.xml 配置 -->
			<property>
			<name>hive.semantic.analyzer.hook</name><value>com.bigdata.hook.TaskIdAuditHook</value>
			</property>自定義 Hook 實現（傳遞 Task ID 到 Ranger 審計）：
Java
public class TaskIdAuditHook implements SemanticAnalyzerHook
			{
			@Override
public ASTNode preAnalyze(ASTNode ast, Context context) {
			    // 從hiveconf獲取任務ID
			    String taskId = HiveConf.getVar(context.getConf(),         HiveConf.ConfVars.HIVESESSIONSPECIFICPARAMS).get("task.id", "unknown");
			    // 設置Ranger審計上下文
			    if (taskId != null && !taskId.isEmpty()) {
			        RangerPerfTracer.logParam("taskId", taskId);
			        // 將Task ID存入ThreadLocal，供Ranger審計插件使用
			        AuditContext.setTaskId(taskId);
			     }
			     return ast;
			     }
			 }2.2 Spark 任務
通過 Spark 配置傳遞 Task ID，並在訪問數據時攜帶：
bash
Bash
			spark-submit \--master yarn \
			  --deploy-mode cluster \--keytab /etc/keytabs/recmd-prod.keytab \--principal recmd-prod@BIGDATA.COM \--conf spark.app.name=recmd-spark-prod-002 \--conf spark.task.id=recmd-spark-prod-002 \--conf spark.hadoop.ranger.audit.taskid=recmd-spark-prod-002 \--class com.data.RecommendationEngine \
			  /opt/jobs/recommendation.jarSpark 讀取 OSS/StarRocks 時傳遞 Task ID：
scala
Scala
			// 讀取OSS數據時攜帶Task ID
val df = spark.read
			  .option("taskId", spark.conf.get("spark.task.id")).parquet("oss://prod-bucket/recommendation/data/")
			  // 寫入StarRocks時攜帶Task ID
			df.write
			  .format("starrocks").option("jdbc.url", "jdbc:mysql://starrocks-fe:9030").option("load-url", "starrocks-fe:8030").option("table.identifier", "prod_db.user_recmd").option("taskId", spark.conf.get("spark.task.id")).save()2.3 Flink 任務
通過 Flink 配置傳遞 Task ID，並在訪問外部系統時攜帶：
bash
Bash
			./bin/flink run-application \
-t yarn-application \
-Dsecurity.kerberos.login.principal=recmd-prod@BIGDATA.COM \
-Dsecurity.kerberos.login.keytab=/etc/keytabs/recmd-prod.keytab \
-Dtask.id=recmd-flink-prod-003 \
-Dflink.ranger.audit.taskid=recmd-flink-prod-003 \
-c com.data.LogProcessor \
			/opt/jobs/log_processor.jarFlink 消費 Kafka 並寫入 OSS 時傳遞 Task ID：
java
運行
Java
			// 配置OSS訪問時的Task ID
			Configuration conf = new Configuration();
			conf.setString("fs.oss.task.id", getRuntimeContext().getExecutionConfig().getGlobalJobParameters().toMap().get("task.id"));
			// 使用帶Task ID的配置創建FileSystem
			FileSystem ossFs = FileSystem.get(new URI("oss://prod-bucket/logs/"), conf);
			// 寫入數據時攜帶Task ID
			DataStream<String> logStream = ...;
			logStream.addSink(new RichSinkFunction<String>() {
			    @Override
			    public void invoke(String value, Context context) throws Exception {
			    // 寫入邏輯中包含Task ID
			        String taskId =     getRuntimeContext().getExecutionConfig().getGlobalJobParameters().toMap().get("task.id");
			        // ...
			        }});三、數據存儲審計日誌的任務標識記錄
3.1 OSS 訪問審計配置
配置 OSS 服務記錄訪問來源的 Task ID：
1. 通過 SDK 傳遞 Task ID：
java
Java
			// OSS客户端配置Task ID
			OSSClientBuilder builder = new OSSClientBuilder();
			ClientConfiguration clientConfig = new ClientConfiguration();
			clientConfig.setUserAgent("taskId:recmd-hive-prod-001");
			// 包含Task ID
			OSS ossClient = builder.build(endpoint, accessKeyId, accessKeySecret, clientConfig);1. 配置 OSS 訪問日誌格式：確保 OSS 訪問日誌包含user-agent字段（其中包含 Task ID）
Plain Text
			[timestamp] [remote_ip] [user_agent] [request_id] [operation] [bucket] [object] [request_size] [response_size] [duration] [status_code]1. 日誌解析規則：從user_agent中提取 Task ID：
regex
Plain Text
			taskId:([a-zA-Z0-9\-]+)3.2 StarRocks 訪問審計配置
配置 StarRocks 記錄查詢請求的 Task ID：
1. 修改 StarRocks FE 配置：
xml
XML
			<!-- fe.conf 配置 -->
			audit_log_level = INFO
			audit_log_file = /var/log/starrocks/fe/audit.log
			audit_log_include_variables = true  # 記錄會話變量1. 查詢時傳遞 Task ID：在 SQL 語句前設置會話變量傳遞 Task ID：
sql
SQL
			-- Hive/Spark/Flink訪問StarRocks時執行
SET @task_id = 'recmd-spark-prod-002';SELECT * FROM prod_db.user_profile WHERE dt = '2023-10-01';1. StarRocks 審計日誌格式：確保審計日誌包含user_vars字段，從中提取task_id
四、Ranger 審計日誌的增強配置
修改 Ranger 配置，確保審計日誌包含 Task ID：
xml
XML
			<!-- ranger-admin-site.xml 配置 -->
			<property>
			    <name>ranger.audit.log4j.appender.AUDITFILE.layout.ConversionPattern</name>
			    <value>{"eventTime":"%d{ISO8601}","reqUser":"%X{username}","accessType":"%X{accessType}",
			  "resource":"%X{resource}","result":"%X{result}","taskId":"%X{taskId}",
			  "clientIP":"%X{clientIP}","sessionId":"%X{sessionId}","requestData":"%X{requestData}"}%n</value>
			  </property>自定義 Ranger 審計插件，從計算引擎上下文提取 Task ID：
java
運行
Java
public class CustomRangerAuditHandler extends RangerDefaultAuditHandler {
			    @Overridepublic void logAudit(RangerAuditEvent auditEvent) {
			    // 從ThreadLocal獲取計算引擎設置的Task ID
			        String taskId = AuditContext.getTaskId();
			        if (taskId != null && !taskId.isEmpty()) {
			            auditEvent.setAdditionalInfo("taskId", taskId);
			            }
			        super.logAudit(auditEvent);
			    }
			}