Spark Streaming：TCP（基本类型）数据源

1 简介

在Spark Streaming的数据源中，TCP的Sockt流和文件流是支持的基本数据流，在官方文档中http://spark.apache.org/docs/1.1.0/streaming-programming-guide.html#input-dstreams，通过如下方式演示了Spark Streaming从TCP接收数据流的情况，如下

- 在一个命令行中执行如下操作

# TERMINAL 1:
# Running Netcat
$ nc -lk 9999
hello world
...

以上命令是将hello world发送到本地端口9999

在另一个命令行中执行如下操作

# TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...`

以上命令是Spark Streaming从本地端口9999接受数据hello word并处理

2 实现自己的nc

nc是系统自带的工具，在此不述及，下面将调用系统的Socket实现自己的TCP数据服务，代码如下

#include <sys/socket.h>
#include <stdlib.h>
#include <arpa/inet.h>
#include <string.h>
#include <stdio.h>
#include <time.h>
#include <unistd.h>

#include <assert.h>
#include <iostream>
using namespace std;
int main(int argc, char** argv)
{
time_t ticks;
char buff[2048];
int socket_id;
//1 定义套接口
socket_id = socket(AF_INET, SOCK_STREAM, 0);
if(socket_id < 0)
{
cout << "socket failed." << endl;
assert(-1);
}
//2 给套接口绑定协议地址
struct sockaddr_in ser_addr, cli_addr;
bzero(&ser_addr, sizeof(ser_addr));
ser_addr.sin_family = AF_INET;
//ser_addr.sin_addr.s_addr = htonl(INADDR_ANY);
inet_aton("localhost",(in_addr*)&ser_addr.sin_addr.s_addr);
ser_addr.sin_port = htons(9999);
int bind_id = bind(socket_id, (struct sockaddr*)&ser_addr, sizeof(ser_addr));
if(bind_id < 0)
{
cout << "bind failed." << endl;
assert(-1);
}
//3 建立监听
int listen_id = listen(socket_id, 1000);
if(listen_id < 0)
{
cout << "listen failed." << endl;
assert(-1);
}
//4 接受连接
int accept_id;
socklen_t cli_len;
while((accept_id = accept(socket_id, (struct sockaddr*)&cli_addr, &cli_len)) > 0)
{
printf("connection from %s, port %d\n", inet_ntop(AF_INET, &cli_addr.sin_addr, buff, sizeof(buff)), ntohs(cli_addr.sin_port));
while(fgets(buff, 2048, stdin))
{
write(accept_id, buff, strlen(buff));
}
close(accept_id);
}
return 0;
}

假设以上代码编译的二进制文件为tcp_source，在命令行中输入如下命令

# TERMINAL 1:
# Running Netcat
$ tcp_source 9999
hello world
...

在另一个命令行启动官方文档中的Demo时，也会得到正确结果，如下

# TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...`