hadoop - mapreduce からハイブ テーブルを読み取る

Question

現在、2 つのハイブ テーブルの違いを見つけるための mapreduce プログラムを作成しています。ハイブ テーブルが 1 つ以上の列に分割されています。したがって、フォルダ名にはパーティション化された列の値が含まれます。

ハイブ パーティション テーブルを読み取る方法はありますか。

マッパーで読めますか？

Answer 1

基礎となる HDFS データは、デフォルトで次のように分割されたハイブ テーブルに編成されるため、

 table/root/folder/x=1/y=1
 table/root/folder/x=1/y=2
 table/root/folder/x=2/y=1
 table/root/folder/x=2/y=2....,

これらの各入力パスをドライバーで作成し、FileInputFormat.addInputPath(job, path) への複数回の呼び出しによってそれらを追加できます。作成したフォルダー パスごとに 1 回の呼び出し。

以下にサンプル コードを貼り付けます。MyMapper.class にパスを追加する方法に注意してください。このサンプルでは、​​MultipleInputs API を使用しています。テーブルは「part」と「xdate」で分割されています。

public class MyDriver extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapred.compress.map.output", "true");
        conf.set("mapred.output.compression.type", "BLOCK"); 

        Job job = new Job(conf);
        //set up various job parameters
        job.setJarByClass(MyDriver.class);
        job.setJobName(conf.get("job.name"));
        MultipleInputs.addInputPath(job, new Path(conf.get("root.folder")+"/xdate="+conf.get("start.date")), TextInputFormat.class, OneMapper.class);
        for (Path path : getPathList(job,conf)) {
            System.out.println("path: "+path.toString());
            MultipleInputs.addInputPath(job, path, Class.forName(conf.get("input.format")).asSubclass(FileInputFormat.class).asSubclass(InputFormat.class), MyMapper.class);
        }
        ...
        ...
        return job.waitForCompletion(true) ? 0 : -2;

    }

    private static ArrayList<Path> getPathList(Job job, Configuration conf) {
        String rootdir = conf.get("input.path.rootfolder");
        String partlist = conf.get("part.list");
        String startdate_s = conf.get("start.date");
        String enxdate_s = conf.get("end.date");
        ArrayList<Path> pathlist = new ArrayList<Path>();
        String[] partlist_split = partlist.split(",");
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date startdate_d = null;
        Date enxdate_d = null;
        Path path = null;
        try {
            startdate_d = sdf.parse(startdate_s);
            enxdate_d = sdf.parse(enxdate_s);
            GregorianCalendar gcal = new GregorianCalendar();
            gcal.setTime(startdate_d);
            Date d = null;
            for (String part : partlist_split) {
                gcal.setTime(startdate_d);
                do {
                    d = gcal.getTime();
                    FileSystem fs = FileSystem.get(conf);
                    path = new Path(rootdir + "/part=" + part + "/xdate="
                            + sdf.format(d));
                    if (fs.exists(path)) {
                        pathlist.add(path);
                    }
                    gcal.add(Calendar.DAY_OF_YEAR, 1);
                } while (d.before(enxdate_d));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return pathlist;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new MyDriver(), args);
        System.exit(res);
    }
}