• 内容
  • 前文回顾
  • 流式处理框架特征
  • 流式处理框架技术需求
  • Spark Streaming 所处地位
  • Spark Streaming 应用实例
  • 结束语
  • 参考资料
  • 评论

在 IBM Bluemix 云平台上开发并部署您的下一个应用。

开始您的试用

public ServerSocket getServerSocket(int port){
ServerSocket server=null;
try {
server = new ServerSocket();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return server;
}
public void sendData(String path,ServerSocket server){
OutputStream out=null;
FileInputStream in=null;
BufferedOutputStream bf =null;
try {
out = server.accept().getOutputStream();
File file = new File(path);
in = new FileInputStream(file);
bf = new BufferedOutputStream(out);
byte[] bt = new byte[(int)file.length()];
in.read(bt);
bf.write(bt);
} catch (IOException e) {
e.printStackTrace();
}finally{
if(in!=null){
try {
in.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(bf!=null){
try {
bf.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(out!=null){
try {
out.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(!server.isClosed()){
try {
server.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
if(args.length<4){
System.err.println("Usage:server3 <port> <file or dir> <send-times> <sleep-time(ms)>");
System.exit(1);
}
Map<Integer, String> fileMap = null;
Server s = new Server();
for (int i = 0; i < Integer.parseInt(args[2]) ; i++) {
ServerSocket server =null;
while(server==null){
server = s.getServerSocket(Integer.parseInt(args[0]));
try {
Thread.sleep(Integer.parseInt(args[3]));
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
while(!server.isBound()){
try {
server.bind(new InetSocketAddress(Integer.parseInt(args[0])));
System.out.println("第"+(i+1)+"个服务端绑定成功");
Thread.sleep(Integer.parseInt(args[3]));
} catch (NumberFormatException | IOException | InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
fileMap = s.getFileMap(args[1]);
System.out.println("fileMap.size="+fileMap.size());
//System.out.println("fileMap="+fileMap);
s.sendData(fileMap.get(s.getNum(0, fileMap.size()-1)), server);
//s.sendData(args[1], server);
}
}
public Map<Integer, String> getMap(String dir,Map<Integer, String> fileMap){
File file = new File(dir);
if(file.isFile()){
if(file.getName().endsWith(".jpg")||file.getName().endsWith(".bmp")|file.getName().
endsWith(".JPG")||file.getName().endsWith(".BMP")){
if(file.length()<1024*1024*2){
fileMap.put(fileMap.size(),file.getAbsolutePath());
}
}else{
}
}
if(file.isDirectory()){
File[] files = file.listFiles();
for (int j = 0; j < files.length; j++) {
getMap(files[j].getAbsolutePath(), fileMap);
}
}
return fileMap;
}
public Map<Integer, String> getFileMap(String dir){
Map<Integer, String> fileMap = new HashMap<Integer, String>();
return getMap(dir, fileMap);
}
public int getNum(int offset,int max){
int i = offset+(int)(Math.random()*max);
if(i>max){
return i-offset;
}else{
return i;
}
}

val s = new SparkConf().setAppName("face")
val sc = new SparkContext(s)
val ssc = new StreamingContext(sc, Seconds(args(0).toInt))
val img = new ImageInputDStream(ssc, args(1), args(2).toInt,
StorageLevel.MEMORY_AND_DISK_SER)//调用重写的 ImageInputDStream 方法读取图片
val imgMap = img.map(x => (new Text(System.currentTimeMillis().toString), x))
imgMap.saveAsNewAPIHadoopFiles("hdfs://spark:9000/image/receiver/img", "", classOf[Text],
classOf[BytesWritable], classOf[ImageFileOutputFormat],
ssc.sparkContext.hadoopConfiguration)//调用 ImageFileOutputFormat 方法写入图片
imgMap.map(x => (x._1, {
if (x._2.getLength > 0) imageModel(x._2) else "-1"
}))//获取 key 的值，即图片
.filter(x => x._2 != "0" && x._2 != "-1")
.map(x => "{time:" + x._1.toString +","+ x._2 + "},").print()
ssc.start()
ssc.awaitTermination()

class ImageInputDStream(@transient ssc_ : StreamingContext,host: String,port:
Int,storageLevel: StorageLevel) extends
ReceiverInputDStream[BytesWritable](ssc_) with Logging{
override def getReceiver(): Receiver[BytesWritable] = {
new ImageRecevier(host,port,storageLevel)
}
}
class ImageRecevier(host: String,port: Int,storageLevel: StorageLevel) extends
Receiver[BytesWritable](storageLevel) with Logging{
override def onStart(): Unit = {
new Thread("Image Socket"){
setDaemon(true)
override def run(): Unit = {
receive()
}
}.start()
}
override def onStop(): Unit = {
}
def receive(): Unit ={
var socket:Socket=null
var in:InputStream =null
try{
logInfo("Connecting to " + host + ":" + port)
socket = new Socket(host, port)
logInfo("Connected to " + host + ":" + port)
in= socket.getInputStream
val buf = new ArrayBuffer[Byte]()
var bytes = new Array[Byte](1024)
var len = 0
while(-1 < len){
len=in.read(bytes)
if(len > 0){
buf ++=bytes
}
}
val bw = new BytesWritable(buf.toArray)
logError("byte:::::"+ bw.getLength)
store(bw)
logInfo("Stopped receiving")
restart("Retrying connecting to " + host + ":" + port)
}catch {
case e: java.net.ConnectException =>
restart("Error connecting to " + host + ":" + port, e)
case t: Throwable =>
restart("Error receiving data", t)
}finally {
if(in!=null){
in.close()
}
if (socket != null) {
socket.close()
logInfo("Closed socket to " + host + ":" + port)
}
}
}

public class ImageFileOutFormat extends FileOutputFormat<Text,BytesWritable> {
@Override
public RecordWriter<Text, BytesWritable> getRecordWriter(TaskAttemptContext taskAttemptContext)
throws IOException, InterruptedException {
Configuration configuration = taskAttemptContext.getConfiguration();
Path path = getDefaultWorkFile(taskAttemptContext, "");
FileSystem fileSystem = path.getFileSystem(configuration);
FSDataOutputStream out = fileSystem.create(path,false);
return new ImageFileRecordWriter(out);
}
protected class ImageFileRecordWriter extends RecordWriter<Text, BytesWritable>{
protected DataOutputStream out;
private final byte[] keyValueSeparator;
private static final String colon=",";
public ImageFileRecordWriter(DataOutputStream out){
this(colon,out);
}
public ImageFileRecordWriter(String keyValueSeparator,DataOutputStream out) {
this.out=out;
this.keyValueSeparator = keyValueSeparator.getBytes();
}
@Override
public void write(Text text, BytesWritable bytesWritable) throws IOException, InterruptedException {
if(bytesWritable!=null){
out.write(bytesWritable.getBytes());
}
}
@Override
public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
out.close();
}
}
}