如何在CDH中安装和使用StreamSets

温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。 Fayson的github：https://github.com/fayson/cdhproject 提示：代码块部分可以左右滑动查看噢

1.关于StreamSets

StreamSets由Informatica前首席产品官Girish Pancha和Cloudera前开发团队负责人Arvind Prabhakar于2014年创立。他们成立该公司主要是应对来自动态数据（data in motion）的挑战 - 包括数据源，数据处理和数据本身，这是一个称为“数据漂移“（https://streamsets.com/reports/data-drift/）的问题。StreamSets设想从头开始管理数据流，避免已有产品和工具的缺陷，并启用一种管理动态数据（data in motion）的新方法。

他们的第一款产品StreamSets Data Collector（https://streamsets.com/products/sdc）被数百家公司用于构建复杂的任意数据流，其中包括财富500强企业，涵盖金融服务，制造业，医疗，媒体，制药和技术等多个行业。

他们最新的产品StreamSets Dataflow Performance Manager，也叫DPM，主要用于构建端到端的数据流。DPM是一个运行控制中心，可以让你映射（数据流），内置的测量和监测确保持续的数据传输和控制动态数据（data in motion）的性能。首先，它将你不同的数据流映射到支持你的每个关键业务流程的拓扑中。然后监测这些拓扑的日常运行情况，根据掌握的性能情况，以满足应用的SLA为目标，确保你始终提供及时和可信的数据。

StreamSets在旧金山，硅谷和巴塞罗那设有办事处，并得到顶级硅谷风险投资公司的支持，包括Accel Partners，Battery Ventures，Ignition Partners和New Enterprise Associates（NEA）。在大数据行业领先的合作伙伴包括Cloudera，Databricks，MapR和Microsoft。

好了，Fayson讲点人话吧，StreamSets是一个大数据采集工具，数据源支持包括结构化和半/非结构化，目标源支持HDFS，HBase，Hive，Kudu，Cloudera Search, ElasticSearch等。它包括一个拖拽式的可视化数据流程设计界面，定时任务调度等功能。举例，它可以将数据源从Kafka+Spark Streaming连接到你的Hadoop集群，而不需要写一行代码。很炫酷有木有！！！下面我们随便看几张截图了再往后走。

2.文档编写目的

本文档主要讲述如何在Cloudera Manager 管理的集群中安装StreamSets和基本使用。

内容概述：

StreamSets安装配置
StreamSets基本使用

3.常见问题

测试环境：

1.操作系统为：Redhat7.2

2.CM和CDH为：5.13.1

3.安装步骤

3.1.安装前准备

到StreamSets官网提供的下载址：https://archives.streamsets.com/index.html

下载以下文件：这里下载使用的是 3.0版本

1.manifest.json 文件

2.STREAMSETS_DATACOLLECTOR-3.0.0.0-el7.parcel文件（比较大约4G左右，包含比较完整的插件）

3.STREAMSETS-3.0.0.0.jar

3.2.配置本地Parcel包

将下载的manifest.json和 STREAMSETS_DATACOLLECTOR-3.0.0.0-el7.parcel，放在一个新建文件夹StreamSets3.0中，并移动到/var/www/html 目录中，做离线包的下载地址，用浏览器访问如下，表示成功

3.3.配置CSD

将STREAMSETS-3.0.0.0.jar拷贝到/opt/cloudera/csd,并更改权限，然后重启cloudera-scm-server服务

[root@cdh4 ~]# cp STREAMSETS-3.0.0.0.jar /opt/cloudera/csd
[root@cdh4 csd]# sudo chown cloudera-scm:cloudera-scm STREAMSETS-3.0.0.0.jar && sudo chmod 644 STREAMSETS-3.0.0.0.jar
[root@cdh4 ~]# systemctl restart cloudera-scm-server

（可左右滑动）

3.4.下载分发激活Parcel包

在CM界面中点击Parcel > 配置 > 添加StreamSets的Parcel包路径，并保持修改

并下载分发和激活(本文档省略该步骤，直接查看激活后的状态)

3.5.添加StreamSets服务

完成上述步骤后，在集群 > 添加服务中添加StreamSets 的服务

选择StreamSets安装

选择一个DataCollector 节点，默认配置，继续等待安装和启动完成

4.StreamSets 的基本使用

官方文档示例：

https://streamsets.com/documentation/datacollector/3.0.0.0/help/index.html#Tutorial/BasicTutorial.html#task_jmz_3dn_ls

4.1.登录

点击 StreamSets > Data Collector WebUI访问

默认的账户: admin 密码:admin

4.2.准备工作

准备工作：

从官网下载测试数据

https://www.streamsets.com/documentation/datacollector/sample_data/tutorial/nyc_taxi_data.csv

创建测试目录并赋予权限：

[root@cdh1 ~]# mkdir /home/test_stream
[root@cdh1 ~]# mkdir /home/test_stream/data
[root@cdh1 ~]# mkdir /home/test_stream/error
[root@cdh1 ~]# mkdir /home/test_stream/out
[root@cdh1 ~]# chmod -R 777 /home/test_stream

（可左右滑动）

将测试数据拷贝到 /home/test_stream/data 目录下

 [root@cdh1 ~]# ll /home/test_stream
 [root@cdh1 ~]# cp nyc_taxi_data.csv /home/test_stream/data/

（可左右滑动）

4.3.创建第一个Pipelines

保存后进入如下页面：

4.4.添加文件目录和修改属性

在侧边中选择Diretory或者Select Origin > Directory ,添加一个目录，选中目录后点击configuration> file,填写Files Directory 和File Name Pattern 还有选择Read Order 为最近更新时间

点击dataFormat 标签，修改选择如下选择

4.5.配置改Pipelines 的错误日志写入方式和目录

4.6.预览文件

将鼠标放到Directory1 上看到只剩一个Validation_0011的异常然后点击预览，Validation_0011仅表示目录原点尚未连接到任何内容，并且不会阻止数据预览

预览配置这里，我们选中预览来源，写入目的地和执行者，显示字段类型，记住配置，然后执行

执行后我们使用表格的方式查看如下：

4.7.添加流选择器

 ${record:value('/payment_type') == 'CRD'}

（可左右滑动）

这里只查看卡类型为信用卡的记录

4.8.使用jython 脚本来筛选卡片

使用Jython脚本创建了一个额外的字段credit_card_type，并通过评估信用卡号码的前几位来生成信用卡类型。如果记录具有信用卡付款类型而没有相应的信用卡号码，则脚本将返回错误消息。（这里下载的完整的parcels包所以Jython插件无需再下载，否则需要在Package Manager 中联网下载）

以下脚本放在 Jython >configuration>Jython>Script 中

try: 
  for record in records:
    cc = record.value['credit_card']
    if cc == '':
      error.write(record, "Payment type was CRD, but credit card was null")
      continue
    cc_type = ''
    if cc.startswith('4'):
      cc_type = 'Visa'
    elif cc.startswith(('51','52','53','54','55')):
      cc_type = 'MasterCard'
    elif cc.startswith(('34','37')):
      cc_type = 'AMEX'
    elif cc.startswith(('300','301','302','303','304','305','36','38')):
      cc_type = 'Diners Club'
    elif cc.startswith(('6011','65')):
      cc_type = 'Discover'
    elif cc.startswith(('2131','1800','35')):
      cc_type = 'JCB'
    else:
      cc_type = 'Other'
    record.value['credit_card_type'] = cc_type
    output.write(record)
except Exception as e:
  error.write(record, e.message)

（可左右滑动）

脚本在评估器中应当如下：