Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
87f54b3e
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
87f54b3e
编写于
2月 03, 2016
作者:
M
Maximilian Michels
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[streaming] remove dummy FlumeSource
上级
4f8b9bb5
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
0 addition
and
198 deletion
+0
-198
flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSource.java
.../apache/flink/streaming/connectors/flume/FlumeSource.java
+0
-149
flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeTopology.java
...pache/flink/streaming/connectors/flume/FlumeTopology.java
+0
-49
未找到文件。
flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSource.java
已删除
100644 → 0
浏览文件 @
4f8b9bb5
///*
// * Licensed to the Apache Software Foundation (ASF) under one or more
// * contributor license agreements. See the NOTICE file distributed with
// * this work for additional information regarding copyright ownership.
// * The ASF licenses this file to You under the Apache License, Version 2.0
// * (the "License"); you may not use this file except in compliance with
// * the License. You may obtain a copy of the License at
// *
// * http://www.apache.org/licenses/LICENSE-2.0
// *
// * Unless required by applicable law or agreed to in writing, software
// * distributed under the License is distributed on an "AS IS" BASIS,
// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// * See the License for the specific language governing permissions and
// * limitations under the License.
// */
//
//package org.apache.flink.streaming.connectors.flume;
//
//import java.util.List;
//
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.functions.source.ConnectorSource;
//import org.apache.flink.streaming.util.serialization.DeserializationSchema;
//import org.apache.flink.util.Collector;
//import org.apache.flume.Context;
//import org.apache.flume.channel.ChannelProcessor;
//import org.apache.flume.source.AvroSource;
//import org.apache.flume.source.avro.AvroFlumeEvent;
//import org.apache.flume.source.avro.Status;
//
//public class FlumeSource<OUT> extends ConnectorSource<OUT> {
// private static final long serialVersionUID = 1L;
//
// String host;
// String port;
// volatile boolean finished = false;
//
// private volatile boolean isRunning = false;
//
// FlumeSource(String host, int port, DeserializationSchema<OUT> deserializationSchema) {
// super(deserializationSchema);
// this.host = host;
// this.port = Integer.toString(port);
// }
//
// public class MyAvroSource extends AvroSource {
// Collector<OUT> output;
//
// /**
// * Sends the AvroFlumeEvent from it's argument list to the Apache Flink
// * {@link DataStream}.
// *
// * @param avroEvent
// * The event that should be sent to the dataStream
// * @return A {@link Status}.OK message if sending the event was
// * successful.
// */
// @Override
// public Status append(AvroFlumeEvent avroEvent) {
// collect(avroEvent);
// return Status.OK;
// }
//
// /**
// * Sends the AvroFlumeEvents from it's argument list to the Apache Flink
// * {@link DataStream}.
// *
// * @param events
// * The events that is sent to the dataStream
// * @return A Status.OK message if sending the events was successful.
// */
// @Override
// public Status appendBatch(List<AvroFlumeEvent> events) {
// for (AvroFlumeEvent avroEvent : events) {
// collect(avroEvent);
// }
//
// return Status.OK;
// }
//
// /**
// * Deserializes the AvroFlumeEvent before sending it to the Apache Flink
// * {@link DataStream}.
// *
// * @param avroEvent
// * The event that is sent to the dataStream
// */
// private void collect(AvroFlumeEvent avroEvent) {
// byte[] b = avroEvent.getBody().array();
// OUT out = FlumeSource.this.schema.deserialize(b);
//
// if (schema.isEndOfStream(out)) {
// FlumeSource.this.finished = true;
// this.stop();
// FlumeSource.this.notifyAll();
// } else {
// output.collect(out);
// }
//
// }
//
// }
//
// MyAvroSource avroSource;
//
// /**
// * Configures the AvroSource. Also sets the output so the application can
// * use it from outside of the invoke function.
// *
// * @param output
// * The output used in the invoke function
// */
// public void configureAvroSource(Collector<OUT> output) {
//
// avroSource = new MyAvroSource();
// avroSource.output = output;
// Context context = new Context();
// context.put("port", port);
// context.put("bind", host);
// avroSource.configure(context);
// // An instance of a ChannelProcessor is required for configuring the
// // avroSource although it will not be used in this case.
// ChannelProcessor cp = new ChannelProcessor(null);
// avroSource.setChannelProcessor(cp);
// }
//
// /**
// * Configures the AvroSource and runs until the user calls a close function.
// *
// * @param output
// * The Collector for sending data to the datastream
// */
// @Override
// public void run(Collector<OUT> output) throws Exception {
// isRunning = true;
// configureAvroSource(output);
// avroSource.start();
// while (!finished && isRunning) {
// this.wait();
// }
// }
//
// @Override
// public void cancel() {
// isRunning = false;
// }
//
//}
flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeTopology.java
已删除
100644 → 0
浏览文件 @
4f8b9bb5
///*
// * Licensed to the Apache Software Foundation (ASF) under one or more
// * contributor license agreements. See the NOTICE file distributed with
// * this work for additional information regarding copyright ownership.
// * The ASF licenses this file to You under the Apache License, Version 2.0
// * (the "License"); you may not use this file except in compliance with
// * the License. You may obtain a copy of the License at
// *
// * http://www.apache.org/licenses/LICENSE-2.0
// *
// * Unless required by applicable law or agreed to in writing, software
// * distributed under the License is distributed on an "AS IS" BASIS,
// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// * See the License for the specific language governing permissions and
// * limitations under the License.
// */
//
//package org.apache.flink.streaming.connectors.flume;
//
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.util.serialization.SerializationSchema;
//import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
//
//public class FlumeTopology {
//
// public static void main(String[] args) throws Exception {
//
// StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
//
// @SuppressWarnings("unused")
// DataStream<String> inputStream1 = env.addSource(
// new FlumeSource<String>("localhost", 41414, new SimpleStringSchema())).addSink(
// new FlumeSink<String>("localhost", 42424, new StringToByteSerializer()));
//
// env.execute();
// }
//
// public static class StringToByteSerializer implements SerializationSchema<String, byte[]> {
//
// private static final long serialVersionUID = 1L;
//
// @Override
// public byte[] serialize(String element) {
// return element.getBytes();
// }
// }
//
//}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录