как создать объект Job при записи набора данных Flink в hdfs - PullRequest
Новая
       5

как создать объект Job при записи набора данных Flink в hdfs

0 голосов
/ 31 октября 2018

Я хочу написать набор данных для улья. Я пробовал куст jdbc, но он не поддерживает batchExecute. Поэтому я решил записать его в hdfs, а затем создать таблицу кустов.

Я пытался использовать следующий код для записи hdfs: 

package test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.hadoop.fs.Path;
import org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.io.IntWritable;
import org.apache.flink.fs.s3presto.shaded.org.apache.hadoop.io.Text;
import org.apache.flink.fs.s3presto.shaded.org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.flink.util.Collector;
import org.apache.hadoop.mapreduce.Job;

public class Test {
    public static void main(String[] args) {

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> text = env.fromElements(
                "Who's there?",
                "I think I hear them. Stand, ho! Who's there?");

            DataSet<Tuple2<String, Integer>> hadoopResult = text
                .flatMap(new LineSplitter())
                .groupBy(0)
                .sum(1);
            //job and jobConf is null,I do not know how to initialize them (new)
        Job job = null;
        Job jobConf = null;
        HadoopOutputFormat<String, Integer> hadoopOF =
                  new HadoopOutputFormat<String, Integer>(
                    new TextOutputFormat<String, Integer>(), job
                  );
                hadoopOF.getConfiguration().set("mapreduce.output.textoutputformat.separator", " ");
                TextOutputFormat.setOutputPath(jobConf, new Path("hdfs://somewhere/"));
                hadoopResult.output(hadoopOF);
    }

     public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 3100297611484689639L;
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                for (String word : line.split(" ")) {
                    out.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        }
}

но есть много ошибок компиляции. Все коды скопированы с официального сайта и прошивая эти коды.

Мой вопрос: Как создать объект Job и Jobconf, а затем записать наборы данных в hdfs?

1 Ответ

0 голосов
/ 01 ноября 2018

Вы найдете полный пример в документации.

Для создания вакансии: 

Job job = Job.getInstance();

Не думаю, что вам нужен объект Jobconf - похоже, вы можете использовать объект Job в обоих местах.

...