2. 程式人生 > >利用streamSet搭建一個簡單的資料管道

利用streamSet搭建一個簡單的資料管道

阿新 • • 發佈：2019-02-15

本教程利用StreamSet搭建一個簡單的資料管道，具體為：從本地目錄中讀取檔案，並在分支中處理讀取的檔案資料，最後將處理的資料寫入到本地目錄。將通過資料預覽來配置資料管道，並新增資料報警功能。

由於是初次嘗試，難免存在錯誤，不足之處還請大神不吝賜教！！！

目錄

0  資料準備

3  資料預覽

10  執行

0  資料準備

/<base directory>/tutorial/origin
/<base directory>/tutorial/destination
/<base directory>/tutorial/error

備註：本教程中的<base directory>為StreamSet的安裝目錄，當然你也可以將資料放在其它位置，只要保證啟動StreamSet的使用者對相應目錄有讀寫許可權

1 配置管道屬性

置管道屬性，主要是對異常資料的處理

  • 在主介面點機 Create New Pipline；

  • 在彈出的對話方塊中輸入 Pipline title 和 optional description，並選擇run the pipeline on Data Collector，點選save進行儲存；

  • 在下面的屬性面板中，選擇Error Records選項，配置Error Records屬性為Write to File(將異常資料儲存到檔案中，這樣就可以在不停止pipeline的情況下處理異常資料)

  •  Write to File的屬性配置如下（沒有的保持預設即可）：

Write to File Property

Description

Directory

Directory for error record files. Enter the directory that you set up for the tutorial. We recommended:

/<base directory>/tutorial/error

Note: To prevent validation errors, the directory must already exist.

Files Prefix

This defines a prefix for error record files.

By default, the files are prefixed with "SDC" and an expression that returns the Data Collector ID, but that's more than we need here.

Delete the default and enter the following prefix: err_

Max File Size

For the tutorial, reduce the files size to something more manageable, such as 5 or 1 MB.

2  配置資料來源（源點）

    資料來源是整個管道的入口，在配置資料來源的時候需要定義連線源點方式、要處理的資料型別以及其它的特殊屬性等

  • 選擇 Select origin -> Directory 或者在最右側的階段庫（Stage Library panel）中選擇Directory origin: .

  • 在下面的屬性面板中選擇File，並配置相關屬性如下（沒有的保持預設即可）：

Directory Property

Value

Files Directory

Directory where you saved the sample file. Enter an absolute path.

We recommended: /<base directory>/tutorial/origin.

File Name Pattern

The Directory origin processes only the files in the directory that match the file name pattern.

The tutorial sample file name is nyc_taxi_data.csv. Since the file is the only file in the directory, you can use something generic, like the asterisk wild card (*) or *.csv.

If you had other .csv files in the directory that you didn't want to process, you might be more specific, like this: nyc_taxi*.csv.

Or if you want to process files with prefixes for other cities, you might use *taxi*.csv.

Read Order

This determines the read order when the directory includes multiple files. You can read based on the last-modified timestamp or file name. Because it's simpler, let's use Last Modified Timestamp.

  • 選擇Data Format配置資料格式屬性如下

Delimited Property

Description

Data Format

The data in the sample file is delimited, so select Delimited.

Delimiter Format Type

Since the sample file is a standard CSV file, use the default: Default CSV (ignores empty lines).

Header Line

The sample file includes a header, so select With Header Line.

Root Field Type

This property determines how the Data Collector processes delimited data. Use the default List-Map.

This allows you to use standard functions to process delimited data. With the List root field type, you need to use delimited data functions.
  • 最後介面看起來像這樣

3  資料預覽

  • 點選預覽圖示(Preview icon） ，按照如下進行預覽配置

Data Preview Property

Description

Preview Source

Use the default Configured Source to use the sample source data.

Write to Destinations and Executors

By default, this property is not selected. In general, you should keep this property clear to avoid writing data to destination systems or triggering tasks in destination systems.

Show Field Type

By default, this property is selected. Keep it selected to see the data type of fields in the record.

Remember the Configuration

Select to use these properties each time you run data preview.

When you select this option, the UI enters data preview without showing this dialog box again.

注：如果無法預覽資料，請檢查前面步驟是否正確，解決除Validation_0011問題以外的其它問題。Validation_0011表示資料沒有任何連線，不影響資料預覽。

4  通過流選擇器路由資料

  • 流選擇器是基於使用者定義的條件規則進行資料過濾，不符合條件的記錄將會被路由到預設的流管道中

  • 選擇Select Processor to Connect > Stream Selector或者在Stage Library panel中選擇 Stream Selectorprocessor ()，然後連線源資料

  • 點選General面板，選擇Required Fields屬性值為/payment_type，如果之前預覽資料沒有問題，則會自動顯示欄位，如果沒有，可以手動輸入

  • 點選condition面板，配置過濾條件，通過點選➕號增加新的條件，滿足過濾條件的將會被分流到管道1，其他的分流到管道2

5  使用Jython處理信用卡型別

  • 通過點選package manage 來 新增Jython Evaluator Processor，點選install即可開始安裝，安裝完成後點選重啟資料收集器

  • 重啟之後就可以選擇Jython Evaluator，將其連線到管道1

  • 點選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)

注：貼上程式碼的時候可能由於格式問題（縮排等）會出錯，請自行調整

  • 出現以下問題又可能安裝Jython的時候沒有安裝成功，缺少了某些庫，建議解除安裝後重新安裝

6 遮蔽信用卡號碼

為了防止信用卡號碼等敏感資訊暴露出來，我們使用欄位掩碼來遮蔽信用卡號碼

  • 在Jython Evaluator之後增加一個Field Mask處理器

  • 然後點選屬性面板中的Mask table，配置如下屬性

Field Masker Property

Configuration

Mask Type

Regular Expression

Regular Expression

(.*)([0-9]{4})

Groups to Show

2

注：正則表示式(.*)([0-9]{4})可以遮蔽除最後四位數字以外的所有數字

7 將資料寫入到目標

在此，我們將資料寫入到本地檔案中

  • 在Field Mask之後連線一個Local FS

  • 點選Output Files 並配置如下屬性值

Local FS Property

Configuration

Files Prefix

Defines a prefix for output file names.

By default, the files are prefixed with "SDC" and an expression that returns the Data Collector ID, but that's more than we need here.

Let's simplify and use "out_" instead.

Directory Template

By default, the directory template includes datetime variables to create a directory structure for output files. This is intended for writing large volumes of data.

Since we only have the sample file to process, we don't need the datetime variables. Go ahead and delete the default and enter the directory where you want the files to be written.

We suggested: /<base directory>/tutorial/destination.

Max File Size (MB)

For the tutorial, let's lower the file size to something manageable, like 5 or 1.

  • 點選Data Format，配置如下：

DELIMITED PROPERTY

CONFIGURATION

Data Format

Delimited

Header Line

With Header Line

8 利用Expression Evaluator新增相應欄位

Jython Evaluator指令碼在信用支付分支中添加了一個新欄位。 為了確保所有記錄具有相同的結構，我們將使用Expression Evaluator將相同的欄位新增到非信用分支。這可確保所有記錄在寫入目標時具有相同的格式。為此，我們使用資料預覽來驗證Jython Evaluator如何將信用卡型別新增到記錄中

  • 點選Preview圖示進行資料預覽，如果此時報錯，請檢查錯誤原因

  • 點選Jython Evaluator即可顯示資料經過Jython Evaluator時的輸入輸出資訊，可以發信output data多了一個欄位credit_card_type.

  • 關閉Preview，增加一個Expression Evaluator連線Stream Selector的第2個管道

  • 點選Expressions並配置如下屬性：

OUTPUT FIELD

EXPRESSION

/credit_card_type

n/a

  • 連線Expression Evaluator到FS destination

9  建立資料規則並報警

現在，在執行基本管道之前，讓我們新增一個數據規則和警報。資料規則是使用者定義的規則，用於檢查在兩個階段之間移動的資料。它們是尋找異常值和異常資料的有力方法。資料規則和警報需要對通過管道的資料進行詳細的理解。對於更一般的管道監測資訊，可以使用度量規則和警報。

  • 在Stream Selector 和 Jython Evaluator, 點選圖示 .然後在下面點選Add

  • 在資料規則對話方塊中配置如下資訊：

DATA RULE PROPERTY

DESCRIPTION

Label

Missing Card Numbers

Condition

${record:value("/credit_card") == ""}

Sampling Percentage

35

Alert Text

At least 10 missing credit card numbers!

Threshold Value

10

  • 點選Save並勾選Active

10  執行

點選開始按鈕即可開啟執行

參考：

相關推薦

利用streamSet搭建一個簡單的資料管道

本教程利用StreamSet搭建一個簡單的資料管道，具體為：從本地目錄中讀取檔案，並在分支中處理讀取的檔案資料，最後將處理的資料寫入到本地目錄。將通過資料預覽來配置資料管道，並新增資料報警功能。 由於是初次嘗試，難免存在錯誤，不足之處還請大神不吝賜教！！！ 目錄 0  

利用red5搭建一個簡單的流媒體直播系統

一、red5安裝、設定。 這個過程就不多說了，參見http://blog.csdn.net/liwf616/article/details/8810405一文,    安裝完成並啟動red5後。 二、安

利用SpringCloud搭建一個最簡單的微服務框架

1.微服務 微服務主要包含服務註冊，服務發現，服務路由，服務配置，服務熔斷，服務降級等一系列的服務，而Spring Cloud為我們提供了個一整套的服務； 本例子為你提供了最簡單的一個服務發現例子，包含服務註冊發現spingCloudEurekaServer、服務配置中

利用js編寫一個簡單的html表單驗證，驗證通過時提交資料(附原始碼)

<!DOCTYPE html> <html lang="en"> <head>     <meta charset="UTF-

【轉】使用webmagic搭建一個簡單的爬蟲

class 直接 了解 pid 個數 粘貼 body 教程 相關配置 【轉】使用webmagic搭建一個簡單的爬蟲 剛剛接觸爬蟲，聽說webmagic很不錯，於是就了解了一下。 webmagic的是一個無須配置、便於二次開發的爬蟲框架，它提供簡單靈活的API，只需少量代碼

搭建LAMP架構— 測試：搭建一個簡單的Discuz論壇

Linux LAMP Discuz 在前面三篇文檔中，我們分別完成了Apache、MySQL、PHP的安裝，現在讓我們玩個遊戲，在已經完成的LAMP上搭建一個簡單的Discuz論壇。軟件包同樣在我之前上傳的百度網盤中。 編譯安裝Apache 編譯安裝MySQL 編譯安裝PHP 百度網盤軟件壓縮包

搭建一個簡單的Eureka程序

stc pen AD ont web容器 pid tap app 但是 Eureka集群主要有三個部分Eureka服務器，服務提供者，服務調用者 簡單的來說就是服務提供者將服務註冊到Eureka服務器，服務調用者對其服務進行查找調用。 Eureka服務程序的搭建可參考官方文

利用meterpreter生成一個簡單的android木馬

是我 linux 學習過程 手機 命令 簡單的 lin 軟件 bubuko 這個學習過程我是在kali linux下進行的，成功的感染了自己的手機，當然了apk包沒改簽名也沒有加小圖標，所以只能當是一個實驗吧。 首先使用這個命令： 使用這個命令，可以生成一個apk包，但是

python網絡編程基礎--socket的簡介,以及使用socket來搭建一個簡單的udp小程序

流程 發送消息 lose 1.10 軟件 搬運 我們 arm 進程間 socket介紹： socket(簡稱套接字)，是進程間通訊的一個工具，他能實現把數據從一方傳輸到另一方，完成不同電腦上進程之間的通訊，它好比數據的搬運工。socket應用:不誇張來說，只要跟網絡相關的應

利用AOP實現一個簡單的快取儲存、清除的工具

基本要求：利用aop實現一個簡單的快取儲存、清除的工具，從實際使用上來說，切面應該在provider層。在service層方法呼叫和資料庫查詢之間生效。為了簡化過程，不要求與資料庫互動，資料可以隨機生成，不要求使用redis等中介軟體，可以直接快取到記憶體中。 程式碼實現非常的基礎，能夠很好

搭建一個簡單的神經網路（向前傳播）

程式碼實現1： #兩層簡單神經網路（全連線） import tensorflow as tf #定義輸入和引數 x=tf.constant([[0.7,0.5]])#一組X，表示體積和重量 w1=tf.Variable(tf.random_normal([2,3],stddev=

使用maven搭建一個簡單的SSM框架

       學習了ssm框架後，一直想要自己親自搭建一個，最近嘗試了一下，效果還行，算是一個微型的ssm框架吧。介紹一下，我使用的IDE是idea。        首先，建立

微服務 ： 使用eclipse 來搭建一個簡單的微服務

下載地址：https://download.csdn.net/download/qq_18430613/10642372 1.先看看整體專案的一個架構，parent 專案父依賴包，用於管理我們的依賴，provider 服務提供方,consumer 消費者。 2.建立parent 專案

利用eureka構建一個簡單的springCloud分散式叢集

1　　通過新建module中的spring initializer新建三個模組,分別是eureka, service-provider, service-consumer 2　　一般自動匯入的依賴就是正確可用的,但是eureka中的spring-cloud-starter-netflix-eureka-cl

使用SpringBoot+Dubbo 搭建一個簡單的分散式服務

Github 地址： github.com/Snailclimb/… ，歡迎各位 Star。 目錄： 使用 SpringBoot+Dubbo 搭建一個簡單分散式服務 實戰之前，先來看幾個重要的概念 什麼是分散式? 什麼是 Duboo

超詳細，新手都能看懂 ！使用SpringBoot+Dubbo 搭建一個簡單的分散式服務

Github 地址：https://github.com/Snailclimb/springboot-integration-examples ，歡迎各位 Star。 目錄： 使用 SpringBoot+Dubbo 搭建一個簡單分散式服務 實戰之前，先來看幾個重要的概念

使用SpringBoot+Dubbo 搭建一個簡單的分散式服務，超詳細，新手都能看懂 ！

使用 SpringBoot+Dubbo 搭建一個簡單分散式服務 實戰之前，先來看幾個重要的概念 開始實戰之前，我們先來簡單的瞭解一下這樣幾個概念：Dubbo、RPC、分散式、由於本文的目的是帶大家使用SpringBoot+Dubbo 搭建一個簡單的分散式服務，所以這些概念我只會簡

利用docker搭建一個git伺服器Gogs

利用docker搭建一個git伺服器Gogs Gogs 是一款極易搭建的自助 Git 圖形化管理介面。 #拉取gogs映象 docker pull gogs/gogs #建立gogs容器 docker run -di --name=gogs -p 10022:22 -p 3000:3

用Python搭建一個簡單的代理池！經理再也不用擔心我的IP被封了

其實每次爬東西的時候，特怕IP被封，所以每次都要把時間延遲設定得長一點... 這次用Python搭建一個簡單的代理池。獲取代理IP，然後驗證其有效性。 不過結果好像不是很理想，為什麼西刺代理的高匿代理都能用？？？ 不是說免費代理不好使嗎？真的是黑人問號臉... / 01 / 代理獲取

ASP.net+MVC2+EasyUI搭建一個簡單表格示例

一個非常基礎的小例子，主要是利用VS2010提供的MVC框架，後臺是ASP.NET，前臺是EasyUI，然後利用EasyUI的datagrid來顯示一個表格的資料。 1.首先建立MVC專案。VS2010自帶MVC2（雖然已經比較老了），所以直接新建一個專案如圖： 生成