我有这样的问题,
{
"name":"someone",
"job":"doctor",
"etc":"etc"
}
在每个json中,对于“job”有不同的值,比如医生、飞行员、司机、看守人等等。我想根据“job”值分离每个json,并将其存储在不同的位置,比如 /home/doctor
, /home/pilot
, /home/driver
等。
我已经尝试使用splitstream函数来实现这一点,但是我必须指定那些值来匹配条件。
public class MyFlinkJob {
private static JsonParser jsonParser = new JsonParser();
private static String key_1 = "doctor";
private static String key_2 = "driver";
private static String key_3 = "pilot";
private static String key_default = "default";
public static void main(String args[]) throws Exception {
Properties prop = new Properties();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties props = new Properties();
props.setProperty("bootstrap.servers", kafka);
props.setProperty("group.id", "myjob");
FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), props);
DataStream<String> record = env.addSource(myConsumer).rebalance()
SplitStream<String> split = record.split(new OutputSelector<String>() {
@Override
public Iterable<String> select(String val) {
JsonObject json = (JsonObject)jsonParser.parse(val);
String jsonValue = CommonFields.getFieldValue(json, "job");
List<String> output = new ArrayList<String>();
if (key_1.equalsIgnoreCase(jsonValue)) {
output.add("doctor");
} else if (key_2.equalsIgnoreCase(jsonValue)) {
output.add("driver");
} else if (key_3.equalsIgnoreCase(jsonValue)) {
output.add("pilot");
} else {
output.add("default");
}
return output;
}});
DataStream<String> doctor = split.select("doctor");
DataStream<String> driver = split.select("driver");
DataStream<String> pilot = split.select("pilot");
DataStream<String> default1 = split.select("default");
doctor.addSink(getBucketingSink(batchSize, prop, key_1));
driver.addSink(getBucketingSink(batchSize, prop, key_2));
pilot.addSink(getBucketingSink(batchSize, prop, key_3));
default1.addSink(getBucketingSink(batchSize, prop, key_default));
env.execute("myjob");
} catch (IOException ex) {
ex.printStackTrace();
} finally {
if (input != null) {
try {
input.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public static BucketingSink<String> getBucketingSink(Long BatchSize, Properties prop, String key) {
BucketingSink<String> sink = new BucketingSink<String>("hdfs://*/home/"+key);
Configuration conf = new Configuration();
conf.set("hadoop.job.ugi", "hdfs");
sink.setFSConfig(conf);
sink.setBucketer(new DateTimeBucketer<String>(prop.getProperty("DateTimeBucketer")));
return sink;
}
}
假设在“job”中有任何其他值,比如engineer或其他什么,而我没有在类中指定它,那么它将转到默认文件夹有没有方法根据“job”的值自动拆分这些json事件而不指定它,并创建一个包含值名的路径,比如/home/enginerr?
1条答案
按热度按时间jyztefdp1#
您需要使用bucketingsink,它支持根据字段的值将记录写入单独的bucket中。我可能有一个map函数,它接收json字符串,解析它,并发出一个
Tuple2<String, String>
,其中第一个元素是job
字段,第二个元素是完整的json字符串。