Kettle 数据集成平台：从桌面客户端到 Web 管理系统

/**
 * 数据集成作业实体
 */
@Data
@TableName("di_job")
public class DiJob {
    /**
     * 作业主键ID
     */
    @TableId(value = "id", type = IdType.AUTO)
    private Long id;

    /**
     * 作业名称
     */
    @NotBlank(message = "作业名称不能为空")
    private String jobName;

    /**
     * 作业类型：TRANSFORMATION(转换) / JOB(作业)
     */
    private String jobType;

    /**
     * Kettle 资源库中的路径
     */
    private String repositoryPath;

    /**
     * 调度表达式 (Cron)
     */
    private String cronExpression;

    /**
     * 作业状态：0-禁用, 1-启用, 2-运行中
     */
    private Integer status;

    /**
     * 创建时间
     */
    private Date createTime;

    /**
     * 更新时间
     */
    private Date updateTime;
}

/**
 * 数据集成作业控制器
 */
@RestController
@RequestMapping("/api/di/job")
public class DiJobController {

    @Autowired
    private DiJobService diJobService;

    /**
     * 分页查询作业列表
     */
    @GetMapping("/page")
    public Result<Page<DiJob>> page(
            @RequestParam(defaultValue = "1") Integer current,
            @RequestParam(defaultValue = "10") Integer size,
            DiJob query) {
        Page<DiJob> page = diJobService.page(
            new Page<>(current, size),
            new QueryWrapper<DiJob>()
                .like(StringUtils.isNotBlank(query.getJobName()),
                      "job_name", query.getJobName())
                .eq(query.getStatus() != null, "status", query.getStatus())
                .orderByDesc("create_time")
        );
        return Result.success(page);
    }

    /**
     * 创建作业
     */
    @PostMapping
    public Result<DiJob> create(@Valid @RequestBody DiJob diJob) {
        diJob.setCreateTime(new Date());
        diJob.setStatus(0); // 默认禁用状态
        diJobService.save(diJob);
        return Result.success(diJob);
    }

    /**
     * 启动作业调度
     */
    @PostMapping("/{id}/start")
    public Result<Void> startJob(@PathVariable Long id) {
        diJobService.startSchedule(id);
        return Result.success();
    }

    /**
     * 停止作业调度
     */
    @PostMapping("/{id}/stop")
    public Result<Void> stopJob(@PathVariable Long id) {
        diJobService.stopSchedule(id);
        return Result.success();
    }

    /**
     * 立即执行作业
     */
    @PostMapping("/{id}/execute")
    public Result<Void> execute(@PathVariable Long id) {
        diJobService.executeOnce(id);
        return Result.success();
    }
}

/**
 * Quartz 调度管理器
 */
@Component
public class QuartzManager {

    @Autowired
    private Scheduler scheduler;

    /**
     * 创建调度任务
     */
    public void createSchedule(Long jobId, String cronExpression) {
        try {
            // 构建任务详情
            JobDetail jobDetail = JobBuilder.newJob(KettleJob.class)
                .withIdentity("JOB_" + jobId, "KETTLE_GROUP")
                .usingJobData("jobId", jobId)
                .build();

            // 构建触发器
            CronTrigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("TRIGGER_" + jobId, "KETTLE_GROUP")
                .withSchedule(CronScheduleBuilder.cronSchedule(cronExpression))
                .build();

            // 注册到调度器
            scheduler.scheduleJob(jobDetail, trigger);

            // 启动调度器（如果未启动）
            if (!scheduler.isStarted()) {
                scheduler.start();
            }
        } catch (SchedulerException e) {
            throw new RuntimeException("创建调度任务失败", e);
        }
    }

    /**
     * 删除调度任务
     */
    public void deleteSchedule(Long jobId) {
        try {
            JobKey jobKey = new JobKey("JOB_" + jobId, "KETTLE_GROUP");
            scheduler.deleteJob(jobKey);
        } catch (SchedulerException e) {
            throw new RuntimeException("删除调度任务失败", e);
        }
    }

    /**
     * 暂停调度任务
     */
    public void pauseSchedule(Long jobId) {
        try {
            JobKey jobKey = new JobKey("JOB_" + jobId, "KETTLE_GROUP");
            scheduler.pauseJob(jobKey);
        } catch (SchedulerException e) {
            throw new RuntimeException("暂停调度任务失败", e);
        }
    }

    /**
     * 恢复调度任务
     */
    public void resumeSchedule(Long jobId) {
        try {
            JobKey jobKey = new JobKey("JOB_" + jobId, "KETTLE_GROUP");
            scheduler.resumeJob(jobKey);
        } catch (SchedulerException e) {
            throw new RuntimeException("恢复调度任务失败", e);
        }
    }
}

/**
 * Kettle 作业执行任务
 */
public class KettleJob implements Job {

    @Override
    public void execute(JobExecutionContext context) {
        // 从 JobDataMap 获取作业ID
        Long jobId = context.getJobDetail().getJobDataMap().getLong("jobId");

        // 执行 Kettle 作业
        executeKettleJob(jobId);
    }

    private void executeKettleJob(Long jobId) {
        // 实现细节：调用 Kettle API 执行作业
        // 包括：资源库连接、作业加载、执行、日志记录等
    }
}

/**
 * Kettle 执行服务
 */
@Service
public class DiKettleService {

    /**
     * 执行作业
     */
    public void executeJob(String jobPath, Map<String, Object> params) {
        try {
            // 初始化 Kettle 环境
            KettleEnvironment.init();

            // 连接资源库
            Repository repository = connectRepository();

            // 加载作业
            JobMeta jobMeta = repository.loadJob(
                jobPath,
                null
            );

            // 创建作业实例
            Job job = new Job(repository, jobMeta);

            // 设置参数
            if (params != null) {
                params.forEach(job::setVariable);
            }

            // 执行作业
            job.start();

            // 等待执行完成
            job.waitUntilFinished();

            // 记录执行日志
            logExecution(job);

        } catch (Exception e) {
            throw new RuntimeException("执行 Kettle 作业失败", e);
        }
    }

    /**
     * 连接 Kettle 资源库
     */
    private Repository connectRepository() throws KettleException {
        // 根据 DB 配置连接资源库
        // 返回 Repository 实例
    }
}

/**
 * 作业执行状态监听器
 */
public class JobExecutionListener {

    /**
     * 开始执行
     */
    public void onJobStart(Long jobId, String executionId) {
        // 更新执行状态为运行中
        // 记录开始时间
    }

    /**
     * 执行完成
     */
    public void onJobComplete(Long jobId, String executionId, boolean success) {
        // 更新执行状态
        // 记录结束时间
        // 计算执行时长
        // 保存执行日志
    }

    /**
     * 执行失败
     */
    public void onJobError(Long jobId, String executionId, Exception e) {
        // 更新执行状态为失败
        // 记录错误信息
        // 触发告警
    }
}

/**
 * 服务注册组件
 */
@Component
public class ServiceRegistry {

    @Value("${dubbo.application.name}")
    private String serviceName;

    @Value("${server.port}")
    private int serverPort;

    @PostConstruct
    public void registerService() {
        // 获取本地 IP
        String localIp = getLocalIp();

        // 构建服务地址
        String serviceAddress = localIp + ":" + serverPort;

        // 注册到 ZooKeeper
        zkClient.createEphemeral(
            "/services/" + serviceName + "/" + serviceAddress
        );
    }
}