綜述

Falcon提供了週期性的將源資料從外部資料庫（資料庫，drop boxes etc）匯入到Hadoop上以及將講過Hadoop運算的資料匯出到外部的資料庫。

在至今所有的版本中，Falcon僅僅支援關係型資料庫（Mysql，Oracle等）講過JDBC作為外部的資料來源。將來的版本可能增加支援其他型別的資料庫。

先決條件

要將資料匯入和資料匯出需要滿足以下的條件：
- Sqoop 1.4.6+
- Oozie 4.2.0+
- 合適的資料庫聯結器

Note
Falcon通過Sqoop進行資料匯入和資料匯出操作，Sqoop將需要合適的資料庫驅動來連線相應的資料庫。必須確保資料庫的jar包被複制到Oozie中的Sqoop的share lib中：

For example, in order to import and export with MySQL, please make sure the latest MySQL connector
*mysql-connector-java-5.1.31.jar+* is copied into oozie's Sqoop share lib

/user/oozie/share/lib/{lib-dir}/sqoop/mysql-connector-java-5.1.31.jar+

where {lib-dir} value varies in oozie deployments.

用法

實體定義和設定

• 資料來源實體

資料來源實體實現了資料庫連線和資料庫憑據驗證的高度抽象。資料來源實體支援了資料庫特定憑據驗證的讀和寫的介面，如果讀和寫的介面不能使用使用的話，它會呼叫預設的憑據進行驗證。通常情況下，資料來源實體將被系統管理員定義。

接下來的例子定義了一個Mysql的資料來源的實體，這個匯入操作將使用Mysql的讀介面（URI：jdbc:mysql://dbhost/test）使用者名稱：import_usr，密碼：sqoop。匯出操作將使用帶有（URI：jdbc:mysql://dbhost/test）的寫介面採用使用者名稱：export_usr和密碼指定在HDFS的 “/user/ambari-qa/password-store/password_write_user”.這個檔案中。

當讀介面和寫介面的憑據不能使用時，將使用預設的憑據The default credential specifies the password using password alias feature available via hadoop credential functionality. 使用者可以建立一個祕鑰使用如下命令：

hadoop credential -create <alias> -provider <provider-path>

where is a string and is a HDFS jceks file，在執行期間 ，the specified alias will be used to look up the password stored encrypted in the jceks hdfs file specified under the providerPath element.

讀和寫的介面可以使管理員分開讀和寫的工作量：

File: mysql-database.xml

  <?xml version="1.0" encoding="UTF-8"?>
      <datasource colo="west-coast" description="MySQL database on west coast" type="mysql" name="mysql-db" xmlns="uri:falcon:datasource:0.1">
          <tags>[email protected], [email protected]</tags>
          <interfaces>
              <!-- ***** read interface ***** -->
              <interface type="readonly" endpoint="jdbc:mysql://dbhost/test">
                  <credential type="password-text">
                      <userName>import_usr</userName>
                      <passwordText>sqoop</passwordFile>
                  </credential>
              </interface>

              <!-- ***** write interface ***** -->
              <interface type="write"  endpoint="jdbc:mysql://dbhost/test">
                  <credential type="password-file">
                      <userName>export_usr</userName>
                     <passwordFile>/user/ambari-qa/password-store/password_write_user</passwordFile>
                  </credential>
              </interface>

              <!-- *** default credential *** -->
              <credential type="password-alias">
                <userName>sqoop2_user</userName>
                <passwordAlias>
                    <alias>sqoop.password.alias</alias>
                    <providerPath>hdfs://namenode:8020/user/ambari-qa/sqoop_password.jceks</providerPath>
                </passwordAlias>
              </credential>

          </interfaces>

          <driver>
              <clazz>com.mysql.jdbc.Driver</clazz>
              <jar>/user/oozie/share/lib/lib_20150721010816/sqoop/mysql-connector-java-5.1.31</jar>
          </driver>
      </datasource>

• 資料集實體

Feed實體可以使使用者在指定去保持資料生命週期和複製資料的同時定義資料匯入和資料匯出的策略，匯入和匯出策略將引用已經定義的資料庫實體去連線和指定特定的資料表。接下來的例子定義了一個具有匯入和匯出策略的feed實體，匯入和匯出都將引用“mysql-db”實體，匯入操作將使用讀介面和匯出操作將使用寫介面，一個Feed例項將每一個小時被建立一次和由於保留策略的設定這個feed策略將被刪除在90天之後。

File: customer_email_feed.xml

 <?xml version="1.0" encoding="UTF-8"?>
      <!--
       A feed representing Hourly customer email data retained for 90 days
       -->
      <feed description="Raw customer email feed" name="customer_feed" xmlns="uri:falcon:feed:0.1">
          <tags>externalSystem=USWestEmailServers,classification=secure</tags>
          <groups>DataImportPipeline</groups>
          <frequency>hours(1)</frequency>
          <late-arrival cut-off="hours(4)"/>
          <clusters>
              <cluster name="primaryCluster" type="source">
                  <validity start="2015-12-15T00:00Z" end="2016-03-31T00:00Z"/>
                  <retention limit="days(90)" action="delete"/>
                  <import>
                      <source name="mysql-db" tableName="simple">
                          <extract type="full">
                              <mergepolicy>snapshot</mergepolicy>
                          </extract>
                          <fields>
                              <includes>
                                  <field>id</field>
                                  <field>name</field>
                              </includes>
                          </fields>
                      </source>
                      <arguments>
                          <argument name="--split-by" value="id"/>
                          <argument name="--num-mappers" value="2"/>
                      </arguments>
                  </import>
                  <export>
                        <target name="mysql-db" tableName="simple_export">
                            <load type="insert"/>
                            <fields>
                              <includes>
                                <field>id</field>
                                <field>name</field>
                              </includes>
                            </fields>
                        </target>
                        <arguments>
                             <argument name="--update-key" value="id"/>
                        </arguments>
                    </export>
              </cluster>
          </clusters>

          <locations>
              <location type="data" path="/user/ambari-qa/falcon/demo/primary/importfeed/${YEAR}-${MONTH}-${DAY}-${HOUR}-${MINUTE}"/>
              <location type="stats" path="/none"/>
              <location type="meta" path="/none"/>
          </locations>

          <ACL owner="ambari-qa" group="users" permission="0755"/>
          <schema location="/none" provider="none"/>

      </feed>

• 匯入策略（Import policy）

匯入策略將使用datasource實體指定特定連線特定的資料來源，表名應該在存在引用的資料庫中，匯入過程可以指定匯入特定的欄位並且可以指定是每次都是全部一次性倒入或者增量匯入，
The merge policy specifies how to organize (snapshot or append, i.e time series partiitons) the data on hadoop.有效的組合是：

1. [full,snapshot] ：資料每次被全部抽取並且被儲存在Feed指定的位置。

2. [incremental, append] ：data is extracted incrementally using the key specified in the deltacolumn and added as a partition to the feed instance location.

3. [incremental, snapshot]. ：資料被增量的抽取和已經在Hadoop中存在的資料進行融合產生一個最新的Feed例項，.這個特性現在並不支援. 這個操作對於一個具有大量的資料的表一以及更新後可以插入到Hadoop的資料裡，對於使用者來說，它可以作為一個整體。

接下來的例子定義了[incremental, append]的策略：

 <import>
                <source name="mysql-db" tableName="simple">
                    <extract type="incremental">
                        <deltacolumn>modified_time</deltacolumn>
                        <mergepolicy>append</mergepolicy>
                    </extract>
                    <fields>
                        <includes>
                            <field>id</field>
                            <field>name</field>
                        </includes>
                    </fields>
                </source>
                <arguments>
                    <argument name="--split-by" value="id"/>
                    <argument name="--num-mappers" value="2"/>
                </arguments>
            </import>

Filed標籤將一定哪一個列將被匯入，預設情況下，所有的列將被匯入，Includes標籤定義了其中需要匯入的列的欄位，Excludes正好與Includes相反，

arguments標籤將可以呼叫在Sqoop中任何額外的引數。

• 匯出策略（Export policy）

匯出策略和匯入策略同理，使用Datasource去連線資料庫，載入特定列的資料對資料表進行插入或者更新資料，Fields操作的選項和Import具有同樣的策略。表名應該存在於資料庫中。

操作步驟：

    ## submit the mysql-db datasource defined in the file mysql_datasource.xml
    falcon entity -submit -type datasource -file mysql_datasource.xml

    ## submit the customer_feed specified in the customer_email_feed.xml
    falcon entity -submit -type feed -file customer_email_feed.xml

    ## schedule the customer_feed
    falcon entity -schedule -type feed -name customer_feed