企业数据切割写入到hdfs

水深无声 2022-10-27 01:40 154阅读 0赞

**目标：将数据按照年月日的结构，将相同日期的数据整理到一个文件。**  
       现有历史数据，一类数据在一个文件中。这个文件中包含了多天、多月、甚至多年的数据。生产系统中需要将这些数据分区存储，即一天一个分区。当天的数据放在当天的分区中。  
       要实现这个功能需要将数据进行拆分。遍历数据中的每一条数据，判断每条数据的所属日期（数据中包含日期格式的数据），将相同日期的数据存放在一个文件中，文件名称以日期为文件名。  
**`实现方法一：`**  
       依次读取数据中的每条数据，获取数据中的日期字段，字段格式为“yyyy-MM-dd HH:mm:ss”。截取出字段中的日期“yyyy-MM-dd”，将相同日期的数据存放在一个文件中，文件名称以日期为文件名。  
**`实现方式二：`**  
       利用MapReduce编程框架“先分后和”实现相同的功能。Map端获取每一条数据，截取出数据中的日期字段，根据日期字段再截取出其中的日期。最后将日期作为key，将原始的一整行数据作为value,进行输出。  
       经过shuffle阶段后数据进入Reduce阶段。reduce阶段获取的数据key 是日期，value List便是这个日期内的所有数据。我们只需每执行一遍reduce ，在执行时使其输在HDFS上创建一个文件，文件名为reduce的key,即数据日期，将reduce的value List写入文件即可。  
**数据**

链接：https://pan.baidu.com/s/1jFauGIhbGuQReeLD6ZcEWQ 
    提取码：1h0y

**数据类型**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]

**创建工程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]  
依赖

<dependencies>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.7.7</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>2.7.7</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>2.7.7</version>
            </dependency>
    
        </dependencies>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.0</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                        
                    </configuration>
                </plugin>
    
                <plugin>
                    
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>2.4.3</version>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <minimizeJar>true</minimizeJar>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
    
            </plugins>
        </build>

下面分别写了mapper reducer driver类

package com.donglin;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class DateSplitMapper extends Mapper<LongWritable, Text, Text, Text> { 
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException { 
            //将数据转换成string
            String linedata = value.toString().trim();
            String[] linedatas = linedata.split("\\t");
    
            //验证数据是否有12个数据(角标是11)
            if (linedatas.length>11 && linedatas[11]!=""){ 
                //2018-01-20 20:18:18
                //获取数据中日期字段(包含时分秒)
                String dateTime = linedatas[11];
    
                //判断datetime是否含有空格，若有则切分数据，获取日期，若没有，那么就跳过
                if (dateTime.contains(" ")){ 
                    //获取日期字段中的日期(包含时分秒)
                    String date = dateTime.substring(0, dateTime.indexOf(" "));
                    //将日期作为key 原始数据为value进行输出
                    context.write(new Text(date),value);
                }
    
            }
    
        }
    }

package com.donglin;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    import java.net.URI;
    
    public class DataSplitReduce extends Reducer <Text,Text,Text,Text>{ 
    
        //key 2018-01-20
        //values 2018-01-20日期内的所有数据
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException { 
    
            Configuration conf = new Configuration();
    
            //对数据的有效性进行验证 若key 包含"-" 表示数据有效，反之无效
            if (key.toString().contains("-")){ 
    
                //1拆分日期数据 key
    
                //年
                String year = key.toString().split("-")[0];
                //月
                String month = key.toString().split("-")[1];
                //日
                String day = key.toString().split("-")[2];
                //使用年月日拼接路径
                Path dfs = new Path("/DeviceDatas/" + year + "/" + month + "/" + day + "/" + key.toString() + ".txt");
    
                //2写入数据需要输出流
                //实例一个文件系统(集群)
                FileSystem hdfs = FileSystem.get(URI.create("hdfs://hadoop102:9000/"), conf);
                FSDataOutputStream outputStream = hdfs.create(dfs);
    
                //3遍历当前key对应的所有数据，将多个数据拼接在一起 写入hdfs 做准备
                String alldatas = "";
                for (Text tmpvalue : values) { 
                    //回车 换行
                    alldatas=alldatas+tmpvalue.toString()+"\r\n";
                }
                //将String类型的数据转换成byte
                byte[] buff = alldatas.getBytes();
    
                //4.将数据写入HDFS
                outputStream.write(buff);
    
                //5关闭数据流
                outputStream.close();
                dfs=null;
                alldatas="";
                buff=null;
            }
    
        }
    }

package com.donglin;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    public class DataSplitDriver { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    
            Configuration configuration = new Configuration();
    
            //1.获取Job
            Job job = Job.getInstance(configuration);
    
            //2.设置Jar包路径
            job.setJarByClass(DataSplitDriver.class);
    
            //3.设置Mapper类输出Kv类型
            job.setMapperClass(DateSplitMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
    
            //4.设置最终输出的KV类型
            job.setReducerClass(DataSplitReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
    
            //5.设置输入输出的路径
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
    
            //6.提交任务
            boolean result = job.waitForCompletion(true);
            System.exit(result?0:1);
        }
    
        }

打包集群上

hadoop jar data-cleaning-1.0-SNAPSHOT.jar com.donglin.DataSplitDriver /historydata/device_db.sql /datacleaning1

解释一下参数  
/historydata/device\_db.sql是hdfs上路径需要指定  
/datacleaning1这个是输出路径，不需要自己创建  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221024/8ad8ca054d804b5db350590869858ff2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: /images/20221024/e862af97ce334437b8070dadf0b47a12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]: /images/20221024/f575f328c99345aeb496f95fba2bba13.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 3]: /images/20221024/81b60621594e4a7783fd77dc346535e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 4]: /images/20221024/0bf831d5f5f34e41a6e9ca9f2043c44c.png