HCatalog - Định dạng đầu ra đầu vào

Các HCatInputFormat và HCatOutputFormatgiao diện được sử dụng để đọc dữ liệu từ HDFS và sau khi xử lý, ghi dữ liệu kết quả vào HDFS bằng công việc MapReduce. Hãy để chúng tôi xây dựng các giao diện định dạng Đầu vào và Đầu ra.

HCatInputFormat

Các HCatInputFormatđược sử dụng với các công việc MapReduce để đọc dữ liệu từ các bảng do HCatalog quản lý. HCatInputFormat hiển thị một API MapReduce Hadoop 0.20 để đọc dữ liệu như thể nó đã được xuất bản lên một bảng.

Sr.No. Tên & Mô tả phương pháp
1

public static HCatInputFormat setInput(Job job, String dbName, String tableName)throws IOException

Đặt đầu vào để sử dụng cho công việc. Nó truy vấn siêu thị với đặc điểm kỹ thuật đầu vào đã cho và tuần tự hóa các phân vùng phù hợp thành cấu hình công việc cho các tác vụ MapReduce.
2

public static HCatInputFormat setInput(Configuration conf, String dbName, String tableName) throws IOException

Đặt đầu vào để sử dụng cho công việc. Nó truy vấn siêu thị với đặc điểm kỹ thuật đầu vào đã cho và tuần tự hóa các phân vùng phù hợp thành cấu hình công việc cho các tác vụ MapReduce.
3

public HCatInputFormat setFilter(String filter)throws IOException

Đặt bộ lọc trên bảng đầu vào.
4

public HCatInputFormat setProperties(Properties properties) throws IOException

Đặt thuộc tính cho định dạng đầu vào.

Các HCatInputFormat API bao gồm các phương thức sau:

  • setInput
  • setOutputSchema
  • getTableSchema

Để sử dụng HCatInputFormat để đọc dữ liệu, trước tiên hãy khởi tạo một InputJobInfo với thông tin cần thiết từ bảng được đọc và sau đó gọi setInput với InputJobInfo.

Bạn có thể dùng setOutputSchema phương pháp bao gồm một projection schema, để chỉ định các trường đầu ra. Nếu một lược đồ không được chỉ định, tất cả các cột trong bảng sẽ được trả về. Bạn có thể sử dụng phương thức getTableSchema để xác định lược đồ bảng cho một bảng đầu vào được chỉ định.

HCatOutputFormat

HCatOutputFormat được sử dụng với các công việc MapReduce để ghi dữ liệu vào các bảng do HCatalog quản lý. HCatOutputFormat đưa ra một API MapReduce Hadoop 0.20 để ghi dữ liệu vào bảng. Khi một công việc MapReduce sử dụng HCatOutputFormat để ghi đầu ra, OutputFormat mặc định được cấu hình cho bảng sẽ được sử dụng và phân vùng mới được xuất bản lên bảng sau khi công việc hoàn thành.

Sr.No. Tên & Mô tả phương pháp
1

public static void setOutput (Configuration conf, Credentials credentials, OutputJobInfo outputJobInfo) throws IOException

Đặt thông tin về đầu ra để ghi cho công việc. Nó truy vấn máy chủ siêu dữ liệu để tìm StorageHandler để sử dụng cho bảng. Nó gây ra lỗi nếu phân vùng đã được xuất bản.
2

public static void setSchema (Configuration conf, HCatSchema schema) throws IOException

Đặt lược đồ cho dữ liệu đang được ghi vào phân vùng. Lược đồ bảng được sử dụng theo mặc định cho phân vùng nếu nó không được gọi.
3

public RecordWriter <WritableComparable<?>, HCatRecord > getRecordWriter (TaskAttemptContext context)throws IOException, InterruptedException

Nhận người viết hồ sơ cho công việc. Nó sử dụng OutputFormat mặc định của StorageHandler để lấy trình ghi bản ghi.
4

public OutputCommitter getOutputCommitter (TaskAttemptContext context) throws IOException, InterruptedException

Nhận trình cam kết đầu ra cho định dạng đầu ra này. Nó đảm bảo rằng đầu ra được cam kết chính xác.

Các HCatOutputFormat API bao gồm các phương thức sau:

  • setOutput
  • setSchema
  • getTableSchema

Cuộc gọi đầu tiên trên HCatOutputFormat phải là setOutput; bất kỳ lệnh gọi nào khác sẽ ném ra một ngoại lệ nói rằng định dạng đầu ra không được khởi tạo.

Lược đồ cho dữ liệu đang được ghi ra được chỉ định bởi setSchemaphương pháp. Bạn phải gọi phương thức này, cung cấp lược đồ dữ liệu bạn đang viết. Nếu dữ liệu của bạn có cùng một giản đồ với giản đồ bảng, bạn có thể sử dụngHCatOutputFormat.getTableSchema() để lấy giản đồ bảng và sau đó chuyển nó cùng với setSchema().

Thí dụ

Chương trình MapReduce sau đây đọc dữ liệu từ một bảng mà nó giả định có một số nguyên trong cột thứ hai ("cột 1") và đếm có bao nhiêu trường hợp của mỗi giá trị riêng biệt mà nó tìm thấy. Nghĩa là, nó tương đương với "select col1, count(*) from $table group by col1;".

Ví dụ: nếu các giá trị trong cột thứ hai là {1, 1, 1, 3, 3, 5}, thì chương trình sẽ tạo ra kết quả sau của các giá trị và số đếm:

1, 3
3, 2
5, 1

Bây giờ chúng ta hãy xem mã chương trình -

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import org.apache.HCatalog.common.HCatConstants;
import org.apache.HCatalog.data.DefaultHCatRecord;
import org.apache.HCatalog.data.HCatRecord;
import org.apache.HCatalog.data.schema.HCatSchema;

import org.apache.HCatalog.mapreduce.HCatInputFormat;
import org.apache.HCatalog.mapreduce.HCatOutputFormat;
import org.apache.HCatalog.mapreduce.InputJobInfo;
import org.apache.HCatalog.mapreduce.OutputJobInfo;

public class GroupByAge extends Configured implements Tool {

   public static class Map extends Mapper<WritableComparable, 
      HCatRecord, IntWritable, IntWritable> {
      int age;
		
      @Override
      protected void map(
         WritableComparable key, HCatRecord value,
         org.apache.hadoop.mapreduce.Mapper<WritableComparable,
         HCatRecord, IntWritable, IntWritable>.Context context
      )throws IOException, InterruptedException {
         age = (Integer) value.get(1);
         context.write(new IntWritable(age), new IntWritable(1));
      }
   }
	
   public static class Reduce extends Reducer<IntWritable, IntWritable,
      WritableComparable, HCatRecord> {
      @Override
      protected void reduce(
         IntWritable key, java.lang.Iterable<IntWritable> values,
         org.apache.hadoop.mapreduce.Reducer<IntWritable, IntWritable,
         WritableComparable, HCatRecord>.Context context
      )throws IOException ,InterruptedException {
         int sum = 0;
         Iterator<IntWritable> iter = values.iterator();
			
         while (iter.hasNext()) {
            sum++;
            iter.next();
         }
			
         HCatRecord record = new DefaultHCatRecord(2);
         record.set(0, key.get());
         record.set(1, sum);
         context.write(null, record);
      }
   }
	
   public int run(String[] args) throws Exception {
      Configuration conf = getConf();
      args = new GenericOptionsParser(conf, args).getRemainingArgs();
		
      String serverUri = args[0];
      String inputTableName = args[1];
      String outputTableName = args[2];
      String dbName = null;
      String principalID = System
		
      .getProperty(HCatConstants.HCAT_METASTORE_PRINCIPAL);
      if (principalID != null)
      conf.set(HCatConstants.HCAT_METASTORE_PRINCIPAL, principalID);
      Job job = new Job(conf, "GroupByAge");
      HCatInputFormat.setInput(job, InputJobInfo.create(dbName, inputTableName, null));

      // initialize HCatOutputFormat
      job.setInputFormatClass(HCatInputFormat.class);
      job.setJarByClass(GroupByAge.class);
      job.setMapperClass(Map.class);
      job.setReducerClass(Reduce.class);
		
      job.setMapOutputKeyClass(IntWritable.class);
      job.setMapOutputValueClass(IntWritable.class);
      job.setOutputKeyClass(WritableComparable.class);
      job.setOutputValueClass(DefaultHCatRecord.class);
		
      HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName, outputTableName, null));
      HCatSchema s = HCatOutputFormat.getTableSchema(job);
      System.err.println("INFO: output schema explicitly set for writing:" + s);
      HCatOutputFormat.setSchema(job, s);
      job.setOutputFormatClass(HCatOutputFormat.class);
      return (job.waitForCompletion(true) ? 0 : 1);
   }
	
   public static void main(String[] args) throws Exception {
      int exitCode = ToolRunner.run(new GroupByAge(), args);
      System.exit(exitCode);
   }
}

Trước khi biên dịch chương trình trên, bạn phải tải xuống một số jars và thêm chúng vào classpathcho ứng dụng này. Bạn cần tải xuống tất cả các lọ Hive và lọ HCatalog (HCatalog-core-0.5.0.jar, hive-di căn-0.10.0.jar, libthrift-0.7.0.jar, hive-executive-0.10.0.jar, libfb303-0.7.0.jar, jdo2-api-2.3-ec.jar, slf4j-api-1.6.1.jar).

Sử dụng các lệnh sau để sao chép những jar tập tin từ local đến HDFS và thêm chúng vào classpath.

bin/hadoop fs -copyFromLocal $HCAT_HOME/share/HCatalog/HCatalog-core-0.5.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/hive-metastore-0.10.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/libthrift-0.7.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/hive-exec-0.10.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/libfb303-0.7.0.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/jdo2-api-2.3-ec.jar /tmp
bin/hadoop fs -copyFromLocal $HIVE_HOME/lib/slf4j-api-1.6.1.jar /tmp

export LIB_JARS=hdfs:///tmp/HCatalog-core-0.5.0.jar,
hdfs:///tmp/hive-metastore-0.10.0.jar,
hdfs:///tmp/libthrift-0.7.0.jar,
hdfs:///tmp/hive-exec-0.10.0.jar,
hdfs:///tmp/libfb303-0.7.0.jar,
hdfs:///tmp/jdo2-api-2.3-ec.jar,
hdfs:///tmp/slf4j-api-1.6.1.jar

Sử dụng lệnh sau để biên dịch và thực thi chương trình đã cho.

$HADOOP_HOME/bin/hadoop jar GroupByAge tmp/hive

Bây giờ, hãy kiểm tra thư mục đầu ra của bạn (hdfs: user / tmp / hive) cho đầu ra (part_0000, part_0001).

ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved