Azure Data Factoryを使用して、MySQLからAzureデータウェアハウスへデータを段階的にインポートする方法？

Question

私はAzure Data Factoryを使用して、定期的にMySQLからAzure SQLデータウェアハウスにデータをインポートしています。

データはAzureストレージアカウントのステージングブロブストレージを通過しますが、パイプラインを実行するとBLOBテキストを列に戻すことができないため失敗します。パイプラインが宛先に挿入しようとする各行は、 "⯑"文字で区切られたすべての列値を含む長い文字列になります。

インクリメンタルメカニズムを試さずにデータファクトリを使用しましたが、正常に動作しました。私はそのような行動を引き起こす理由は見当たりませんが、おそらく何かが欠けているでしょう。

私は、パイプラインを説明するJSONをマイナーチェンジしています。これを説明できるものがあれば教えてください。

ありがとうございます！

EDIT：追加例外メッセージ：

Failed execution Database operation failed. Error message from database execution : ErrorCode=FailedDbOperation,'Type=Microsoft.DataTransfer.Com‌​mon.Shared.HybridDel‌​iveryException,Messa‌​ge=Error happened when loading data into SQL Data Warehouse.,Source=Microsoft.DataTransfer.ClientLibrary,''Typ‌​e=System.Data.SqlCli‌​ent.SqlException,Mes‌​sage=Query aborted-- the maximum reject threshold (0 rows) was reached while reading from an external source: 1 rows rejected out of total 1 rows processed. (/f4ae80d1-4560-4af9-9e74-05de941725ac/Data.8665812f-fba1-40‌​7a-9e04-2ee5f3ca5a7e‌​.txt) Column ordinal: 27, Expected data type: VARCHAR(45) collate SQL_Latin1_General_CP1_CI_AS, Offending value:* ROW OF VALUES * (Tokenization failed), Error: Not enough columns in this line.,},],'.

{ 
"name": "CopyPipeline-move_incremental_test", 
"properties": { 
    "activities": [ 
     { 
      "type": "Copy", 
      "typeProperties": { 
       "source": { 
        "type": "RelationalSource", 
        "query": "$$Text.Format('select * from [table] where InsertTime >= \\'{0:yyyy-MM-dd HH:mm}\\' AND InsertTime < \\'{1:yyyy-MM-dd HH:mm}\\'', WindowStart, WindowEnd)" 
       }, 
       "sink": { 
        "type": "SqlDWSink", 
        "sqlWriterCleanupScript": "$$Text.Format('delete [schema].[table] where [InsertTime] >= \\'{0:yyyy-MM-dd HH:mm}\\' AND [InsertTime] <\\'{1:yyyy-MM-dd HH:mm}\\'', WindowStart, WindowEnd)", 
        "allowPolyBase": true, 
        "polyBaseSettings": { 
         "rejectType": "Value", 
         "rejectValue": 0, 
         "useTypeDefault": true 
        }, 
        "writeBatchSize": 0, 
        "writeBatchTimeout": "00:00:00" 
       }, 
       "translator": { 
        "type": "TabularTranslator", 
        "columnMappings": "column1:column1,column2:column2,column3:column3" 
       }, 
       "enableStaging": true, 
       "stagingSettings": { 
        "linkedServiceName": "StagingStorage-somename", 
        "path": "somepath" 
       } 
      }, 
      "inputs": [ 
       { 
        "name": "InputDataset-input" 
       } 
      ], 
      "outputs": [ 
       { 
        "name": "OutputDataset-output" 
       } 
      ], 
      "policy": { 
       "timeout": "1.00:00:00", 
       "concurrency": 10, 
       "style": "StartOfInterval", 
       "retry": 3, 
       "longRetry": 0, 
       "longRetryInterval": "00:00:00" 
      }, 
      "scheduler": { 
       "frequency": "Hour", 
       "interval": 1 
      }, 
      "name": "Activity-0-_Custom query_->[schema]_[table]" 
     } 
    ], 
    "start": "2017-06-01T05:29:12.567Z", 
    "end": "2099-12-30T22:00:00Z", 
    "isPaused": false, 
    "hubName": "datafactory_hub", 
    "pipelineMode": "Scheduled" 
} 

}

Answer 1

それはあなたのやっ右が何であるかのように聞こえるが、データが不完全に形成されている（一般的な問題、なしUTF-8エンコーディング） ADFは必要に応じて構造体を解析できません。私がこれに遭遇すると、パイプラインにカスタムアクティビティを追加して、データを整理して準備しなければならないことがよくあります。そのため、下流のアクティビティによって構造化された方法で使用できます。残念ながら、これはソリューションの開発における大きなオーバーヘッドであり、データ変換に対処するためにC＃クラスを作成する必要があります。

また、ADFには独自のコンピューティングがなく、他のサービスのみが呼び出されるため、コンパイル済みコードに対して実行するAzureバッチ・サービスも必要です。

悲しいことに、ここには魔法の修正はありません。 Azureは完璧に構造化されたデータを抽出して読み込むのに最適ですが、現実世界では変換やクリーニングを行うためにETLやETLが必要なパイプラインが必要です。

は、ここであなたが始めるために作成ADFカスタムアクティビティのリンクです：https://www.purplefrogsystems.com/paul/2016/11/creating-azure-data-factory-custom-activities/

・ホープ、このことができます。

Answer 2

私はAzureのSQLデータベースからAzure DWHにステージング（これはPolybaseを意味します）を使用してData Factory v.2を使用してインポートするときに、同じメッセージで苦労しました。ポリベースがSQLから直接使用されていなくてもデータファクトリを介して使用されていても、私が受け取ったメッセージはhereと非常に似ています。とにかく、私の解決策は、10進数または数値型の列のNULL値を避けることでした。たとえば、次のようになりました。 ISNULL（mynumericCol、0）をmynumericColとして返します。