hdfs客户端定时日志采集任务的开发

idea已经新建好了项目，也添加了依赖项，然后要如何用代码来完成这个项目，求解答
工程开发1：新建子包hdfsClient，完成hdfs客户端定时日志采集任务的开发；
任务a：新建LogsTimingCollections类，实现定时日志采集任务及其调度；
任务b：新建LogsTimingCollectionsTask类，实现日志采集线程任务；
任务c：调度定时日志采集任务并执行，观察控制台输出日志

来自GPT的回答：

下面是一个示例代码，展示了如何使用 Java 开发 HDFS 客户端定时日志采集任务：

LogsTimingCollections.java

java
Copy code
package hdfsClient;

import java.util.Timer;
import java.util.TimerTask;

public class LogsTimingCollections {
    public static void main(String[] args) {
        Timer timer = new Timer();
        TimerTask task = new LogsTimingCollectionsTask();

        // 设置定时任务的延迟和周期
        long delay = 0; // 延迟 0 毫秒开始执行
        long period = 5000; // 每隔 5000 毫秒执行一次

        // 执行定时任务
        timer.schedule(task, delay, period);
    }
}
LogsTimingCollectionsTask.java

java
Copy code
package hdfsClient;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimerTask;

public class LogsTimingCollectionsTask extends TimerTask {
    private static final String SOURCE_DIRECTORY = "path/to/source/directory";
    private static final String DESTINATION_DIRECTORY = "path/to/destination/directory";
    private static final String HDFS_URI = "hdfs://localhost:9000";

    @Override
    public void run() {
        try {
            // 获取当前时间作为日志文件名
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd_HHmmss");
            String currentTime = dateFormat.format(new Date());

            // 构建文件路径
            String sourceFilePath = SOURCE_DIRECTORY + File.separator + "logs.txt";
            String destinationFilePath = DESTINATION_DIRECTORY + File.separator + currentTime + ".txt";

            // 复制日志文件到目标目录
            Files.copy(new File(sourceFilePath).toPath(), new File(destinationFilePath).toPath(),
                    StandardCopyOption.REPLACE_EXISTING);

            // 将日志文件上传到 HDFS
            Configuration configuration = new Configuration();
            configuration.set("fs.defaultFS", HDFS_URI);
            FileSystem fileSystem = FileSystem.get(configuration);
            fileSystem.copyFromLocalFile(new Path(destinationFilePath), new Path("/logs/" + currentTime + ".txt"));

            System.out.println("日志采集成功：" + destinationFilePath);
        } catch (IOException e) {
            System.out.println("日志采集失败：" + e.getMessage());
        }
    }
}
以上代码假设您已经有一个源目录（SOURCE_DIRECTORY）包含名为 logs.txt 的日志文件，以及一个目标目录（DESTINATION_DIRECTORY）用于存储采集的日志文件。您需要将这些路径替换为实际的路径。另外，还需要指定 HDFS 的 URI（HDFS_URI）。

在定时任务中，每隔一定时间（此处为 5000 毫秒）执行一次任务，任务的逻辑是将源目录下的 logs.txt 文件复制到目标目录，并将其上传到 HDFS 的 /logs 目录下，同时在控制台输出日志。

请根据您的实际需求进行必要的修改和调整。同时，确保项目中已添加正确的 Hadoop 和 HDFS 客户端依赖项。

工程开发2：新建子包MRClient，完成日志数据的分布式并行计算任务1；
工程任务描述：现大数据集群已采集到一批订单交易数据，如前文图所示；
现编写MR任务，实现需求：统计当前日志数据中记录到的所有产生订单的交易金额的总额；
a)任务d：新建priceCountBean对象，实现一个自定义序列化类型，自行思考类型所需封装数据，构建出自定义序列化bean类型对象；
b)任务e：新建priceCountMapper对象，实现MapTask任务；
c)任务f：新建priceCountReducer对象，实现ReduceTask任务；
d)任务g：新建priceCountDriver对象，实现Job任务对象；
e)任务h：将MR任务打jar包，上传至yarn集群运行；