给出上述例子的完整代码

以下是一个简单的 Hadoop 分析股票数据的完整代码：

1. 数据收集

我们可以从 Yahoo Finance 网站获取 Apple 公司股票历史数据，保存到本地文件中。

2. 数据清洗

在本例中，我们只需要去除无效数据，因此可以手动进行清洗。将清洗后的数据保存到本地文件中。

3. 数据存储

我们将清洗后的数据存储到 Hadoop 分布式文件系统（HDFS）中。

# 将本地文件上传到 HDFS
hdfs dfs -put cleaned_stock_data.csv /user/hadoop/stock_data

4. 数据分析

我们编写一个 MapReduce 程序，计算每天的开盘价和收盘价的平均值，并将结果存储到 HDFS 中。

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class StockAnalysis {

  public static class StockMapper
       extends Mapper<Object, Text, Text, FloatWritable>{

    private Text date = new Text();
    private FloatWritable openPrice = new FloatWritable();
    private FloatWritable closePrice = new FloatWritable();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      String[] fields = value.toString().split(",");
      if (fields.length == 7) {
        date.set(fields[0]);
        openPrice.set(Float.parseFloat(fields[1]));
        closePrice.set(Float.parseFloat(fields[4]));
        context.write(date, openPrice);
        context.write(date, closePrice);
      }
    }
  }

  public static class StockReducer
       extends Reducer<Text,FloatWritable,Text,FloatWritable> {
    private FloatWritable result = new FloatWritable();

    public void reduce(Text key, Iterable<FloatWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      float sum = 0;
      int count = 0;
      for (FloatWritable val : values) {
        sum += val.get();
        count++;
      }
      result.set(sum / count);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "stock analysis");
    job.setJarByClass(StockAnalysis.class);
    job.setMapperClass(StockMapper.class);
    job.setCombinerClass(StockReducer.class);
    job.setReducerClass(StockReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FloatWritable.class);
    FileInputFormat.addInputPath(job, new Path("/user/hadoop/stock_data"));
    FileOutputFormat.setOutputPath(job, new Path("/user/hadoop/stock_analysis"));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

5. 数据可视化

我们可以使用 Tableau 或 Excel 等工具将结果可视化。

以上是一个简单的 Hadoop 分析股票数据的完整代码。要注意的是，实际应用中可能需要更复杂的数据清洗和分析步骤，以及更高级的数据可视化工具和技术。