Доступ к данным из REST API в HIVE

Question

Есть ли способ создать таблицу улья, где местом для этой таблицы улья будет http JSON REST API? Я не хочу импортировать данные каждый раз в HDFS.

Answer 1

Я столкнулся с аналогичной ситуацией в проекте пару лет назад. Это своего рода сдержанный способ проглатывания данных из Restful to HDFS, а затем вы используете аналитику Hive для реализации бизнес-логики. Надеюсь, вы знакомы с ядром Java, Map Reduce (если вы не можете взглянуть на поток данных Hortonworks, HDF, который является продуктом Hortonworks).

Шаг 1: Рабочий процесс обработки данных не должен быть привязан к рабочему процессу Hive, который содержит бизнес-логику. Это должно выполняться независимо в соответствии с вашим требованием (объем скорость потока данных) и регулярно контролироваться. Я пишу этот код в текстовом редакторе. WARN: Это не скомпилировано или не проверено !!

Код, приведенный ниже, использует Mapper, который принимает URL-адрес или настраивает его, чтобы принять список URL-адресов из FS. Полезная нагрузка или запрошенные данные сохраняются в виде текстового файла в указанном каталоге вывода задания (забудьте структуру данных на этот раз).

Mapper Класс:

import java.io.IOException; 
import java.io.InputStream; 
import java.io.OutputStream; 
import java.net.URL; 
import java.net.URLConnection; 

import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IOUtils; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 


public class HadoopHttpClientMap extends Mapper<LongWritable, Text, Text, Text> { 
    private int file = 0; 
    private String jobOutDir; 
    private String taskId; 

    @Override 
    protected void setup(Context context) throws IOException,InterruptedException { 
     super.setup(context); 

     jobOutDir = context.getOutputValueClass().getName(); 
     taskId = context.getJobID().toString(); 

    } 

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{ 

     Path httpDest = new Path(jobOutDir, taskId + "_http_" + (file++)); 

     InputStream is = null; 
     OutputStream os = null; 
     URLConnection connection; 
     try { 
      connection = new URL(value.toString()).openConnection(); 
      //implement connection timeout logics 
      //authenticate.. etc 
      is = connection.getInputStream(); 

      os = FileSystem.getLocal(context.getConfiguration()).create(httpDest,true); 

      IOUtils.copyBytes(is, os, context.getConfiguration(), true); 

     } catch(Throwable t){ 
      t.printStackTrace(); 
     }finally { 
      IOUtils.closeStream(is); 
      IOUtils.closeStream(os); 
     } 

     context.write(value, null); 
     //context.write(new Text (httpDest.getName()), new Text (os.toString())); 
    } 

} 

Mapper Только Работа:

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.FileSystem; 
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.input.TextInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 


public class HadoopHttpClientJob { 
    private static final String data_input_directory = “YOUR_INPUT_DIR”; 
    private static final String data_output_directory = “YOUR_OUTPUT_DIR”; 

    public HadoopHttpClientJob() { 
    } 

    public static void main(String... args) { 
     try { 
      Configuration conf = new Configuration(); 

      Path test_data_in = new Path(data_input_directory, "urls.txt"); 
      Path test_data_out = new Path(data_output_directory); 

      @SuppressWarnings("deprecation") 
      Job job = new Job(conf, "HadoopHttpClientMap" + System.currentTimeMillis()); 
      job.setJarByClass(HadoopHttpClientJob.class); 

      FileSystem fs = FileSystem.get(conf); 

      fs.delete(test_data_out, true); 
      job.setMapperClass(HadoopHttpClientMap.class); 
      job.setMapOutputKeyClass(Text.class); 
      job.setMapOutputValueClass(Text.class); 
      job.setInputFormatClass(TextInputFormat.class); 
      job.setOutputFormatClass(TextOutputFormat.class); 
      job.setNumReduceTasks(0); 

      FileInputFormat.addInputPath(job, test_data_in); 
      FileOutputFormat.setOutputPath(job, test_data_out); 

      job.waitForCompletion(true); 

     }catch (Throwable t){ 
      t.printStackTrace(); 
     } 
    } 
} 

Шаг 2: Создание внешней таблицы в улье на основе каталога HDFS. Не забудьте использовать Hive SerDe для данных JSON (в вашем случае), тогда вы можете скопировать данные из внешней таблицы в управляемые основные таблицы. Это шаг, на котором вы реализуете свои инкрементные логики, сжатие.

Шаг 3: Поместите ваши запросы на улей (которые вы, возможно, уже создали) на главную таблицу, чтобы реализовать ваши бизнес-потребности.

Примечание. Если вы предположительно ссылаетесь на анализ в реальном времени или потоковое использование api, возможно, вам придется изменить архитектуру вашего приложения. Поскольку вы задали архитектурный вопрос, я использую свое лучшее обоснованное предположение, чтобы поддержать вас. Пожалуйста, пройдите через это один раз. Если вы чувствуете, что можете реализовать это в своем приложении, тогда вы можете задать конкретный вопрос, я постараюсь изо всех сил обратиться к ним.