org.apache.flink.api.common.functions.FlatMapFunction类的使用及代码示例

本文整理了Java中org.apache.flink.api.common.functions.FlatMapFunction类的一些代码示例，展示了FlatMapFunction类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。FlatMapFunction类的具体详情如下：
包路径：org.apache.flink.api.common.functions.FlatMapFunction
类名称：FlatMapFunction

FlatMapFunction介绍

[英]Base interface for flatMap functions. FlatMap functions take elements and transform them, into zero, one, or more elements. Typical applications can be splitting elements, or unnesting lists and arrays. Operations that produce multiple strictly one result element per input element can also use the MapFunction.

The basic syntax for using a FlatMapFunction is as follows:

DataSet input = ...;

[中]flatMap函数的基本接口。FlatMap函数获取元素并将其转换为零、一个或多个元素。典型的应用程序可以是拆分元素，或者取消列表和数组的测试。每个输入元素生成多个结果元素的操作也可以使用MapFunction。
使用FlatMapFunction的基本语法如下：

DataSet input = ...;

代码示例

代码示例来源：origin: apache/flink

@Override
public void flatMap(PyObject value, Collector out) throws Exception {
this.collector.setCollector(out);
try {
this.fun.flatMap(value, this.collector);
} catch (PyException pe) {
throw createAndLogException(pe);
}
}
}

代码示例来源：origin: apache/flink

@Override
public void processElement(StreamRecord element) throws Exception {
collector.setTimestamp(element);
userFunction.flatMap(element.getValue(), collector);
}
}

代码示例来源：origin: apache/flink

@Override
protected List executeOnCollections(List inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = this.userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, this.parameters);

ArrayList result = new ArrayList(inputData.size());
ListCollector collector = new ListCollector(result);
for (T element : inputData) {
function.flatMap(element, collector);
}

FunctionUtils.closeFunction(function);

return result;
}
}

代码示例来源：origin: apache/flink

@Override
public void processElement(StreamRecord> element) throws Exception {
userFunction.flatMap(element.getValue(), new TimestampedCollector<>(output));
}

代码示例来源：origin: apache/flink

@Override
protected List executeOnCollections(List input, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, parameters);
ArrayList result = new ArrayList(input.size());
TypeSerializer inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
TypeSerializer outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
CopyingListCollector resultCollector = new CopyingListCollector(result, outSerializer);
for (IN element : input) {
IN inCopy = inSerializer.copy(element);
function.flatMap(inCopy, resultCollector);
}
FunctionUtils.closeFunction(function);
return result;
}
}

代码示例来源：origin: apache/flink

@Override
public void processElement(StreamRecord> element) throws Exception {
userFunction.flatMap(element.getValue(), new TimestampedCollector<>(output));
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

@Override
public void collect(IT record) {
try {
this.numRecordsIn.inc();
this.mapper.flatMap(record, this.outputCollector);
} catch (Exception ex) {
throw new ExceptionInChainedStubException(this.taskName, ex);
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java

@Override
public void processElement(StreamRecord element) throws Exception {
collector.setTimestamp(element);
userFunction.flatMap(element.getValue(), collector);
}
}

代码示例来源：origin: org.apache.flink/flink-runtime

@Override
public void collect(IT record) {
try {
this.numRecordsIn.inc();
this.mapper.flatMap(record, this.outputCollector);
} catch (Exception ex) {
throw new ExceptionInChainedStubException(this.taskName, ex);
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.11

@Override
public void processElement(StreamRecord element) throws Exception {
collector.setTimestamp(element);
userFunction.flatMap(element.getValue(), collector);
}
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

@Override
public void collect(IT record) {
try {
this.numRecordsIn.inc();
this.mapper.flatMap(record, this.outputCollector);
} catch (Exception ex) {
throw new ExceptionInChainedStubException(this.taskName, ex);
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.10

@Override
public void processElement(StreamRecord element) throws Exception {
collector.setTimestamp(element);
userFunction.flatMap(element.getValue(), collector);
}
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

@Override
public void collect(IT record) {
try {
this.numRecordsIn.inc();
this.mapper.flatMap(record, this.outputCollector);
} catch (Exception ex) {
throw new ExceptionInChainedStubException(this.taskName, ex);
}
}

代码示例来源：origin: com.alibaba.blink/flink-core

@Override
protected List executeOnCollections(List inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = this.userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, this.parameters);

ArrayList result = new ArrayList(inputData.size());
ListCollector collector = new ListCollector(result);
for (T element : inputData) {
function.flatMap(element, collector);
}

FunctionUtils.closeFunction(function);

return result;
}
}

代码示例来源：origin: org.apache.flink/flink-core

@Override
protected List executeOnCollections(List inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = this.userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, this.parameters);

ArrayList result = new ArrayList(inputData.size());
ListCollector collector = new ListCollector(result);
for (T element : inputData) {
function.flatMap(element, collector);
}

FunctionUtils.closeFunction(function);

return result;
}
}

代码示例来源：origin: com.alibaba.blink/flink-core

@Override
protected List executeOnCollections(List input, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, parameters);
ArrayList result = new ArrayList(input.size());
TypeSerializer inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
TypeSerializer outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
CopyingListCollector resultCollector = new CopyingListCollector(result, outSerializer);
for (IN element : input) {
IN inCopy = inSerializer.copy(element);
function.flatMap(inCopy, resultCollector);
}
FunctionUtils.closeFunction(function);
return result;
}
}

代码示例来源：origin: org.apache.flink/flink-core

@Override
protected List executeOnCollections(List input, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
FlatMapFunction function = userFunction.getUserCodeObject();

FunctionUtils.setFunctionRuntimeContext(function, ctx);
FunctionUtils.openFunction(function, parameters);
ArrayList result = new ArrayList(input.size());
TypeSerializer inSerializer = getOperatorInfo().getInputType().createSerializer(executionConfig);
TypeSerializer outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
CopyingListCollector resultCollector = new CopyingListCollector(result, outSerializer);
for (IN element : input) {
IN inCopy = inSerializer.copy(element);
function.flatMap(inCopy, resultCollector);
}
FunctionUtils.closeFunction(function);
return result;
}
}

代码示例来源：origin: vasia/gelly-streaming

f.flatMap(input1, out2);
Assert.assertEquals(0, resultList2.size());
f.flatMap(input2, out2);
Assert.assertEquals(0, resultList2.size());
f.flatMap(input3, out2);
Assert.assertEquals(5, resultList2.size());
Assert.assertEquals(true, resultList2.contains(new Tuple2<>(3, 1)));
f.flatMap(input4, out2);
Assert.assertEquals(5, resultList2.size());
Assert.assertEquals(true, resultList2.contains(new Tuple2<>(2, 1)));

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

@Override
public void run() throws Exception {
final Counter numRecordsIn = this.taskContext.getMetricGroup().getIOMetricGroup().getNumRecordsInCounter();
final Counter numRecordsOut = this.taskContext.getMetricGroup().getIOMetricGroup().getNumRecordsOutCounter();
// cache references on the stack
final MutableObjectIterator input = this.taskContext.getInput(0);
final FlatMapFunction function = this.taskContext.getStub();
final Collector output = new CountingCollector<>(this.taskContext.getOutputCollector(), numRecordsOut);
if (objectReuseEnabled) {
IT record = this.taskContext.getInputSerializer(0).getSerializer().createInstance();
while (this.running && ((record = input.next(record)) != null)) {
numRecordsIn.inc();
function.flatMap(record, output);
}
} else {
IT record;
while (this.running && ((record = input.next()) != null)) {
numRecordsIn.inc();
function.flatMap(record, output);
}
}
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

@Override
public void run() throws Exception {
final Counter numRecordsIn = this.taskContext.getMetricGroup().getIOMetricGroup().getNumRecordsInCounter();
final Counter numRecordsOut = this.taskContext.getMetricGroup().getIOMetricGroup().getNumRecordsOutCounter();
// cache references on the stack
final MutableObjectIterator input = this.taskContext.getInput(0);
final FlatMapFunction function = this.taskContext.getStub();
final Collector output = new CountingCollector<>(this.taskContext.getOutputCollector(), numRecordsOut);
if (objectReuseEnabled) {
IT record = this.taskContext.getInputSerializer(0).getSerializer().createInstance();
while (this.running && ((record = input.next(record)) != null)) {
numRecordsIn.inc();
function.flatMap(record, output);
}
} else {
IT record;
while (this.running && ((record = input.next()) != null)) {
numRecordsIn.inc();
function.flatMap(record, output);
}
}
}