A sample AWS DataPipeline configuration which loads CSV data from S3 to DynamoDB

CSV2DynamoDBDataPipeline.json

{
  "objects": [
    {
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "writeThroughputPercent": "0.95",
      "dataFormat": {
        "ref": "DynamoDBDataFormatId_suGD7"
      },
      "name": "OutputDynamoDBTable",
      "id": "DataNodeId_h3tjF",
      "type": "DynamoDBDataNode",
      "region": "eu-west-1",
      "tableName": "test-import-table"
    },
    {
      "name": "InputCSVFormat",
      "column": [
        "col_a string",
        "col_b string",
        "col_c string",
        "col_d string",
        "col_e string",
        "col_f string",
        "col_g string",
        "col_h bigint",
        "col_i bigint"
      ],
      "id": "DataFormatId_yYuAT",
      "type": "CSV"
    },
    {
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "directoryPath": "s3://dynamodb-imports/test",
      "dataFormat": {
        "ref": "DataFormatId_yYuAT"
      },
      "name": "InputS3Data",
      "id": "DataNodeId_KlnIq",
      "type": "S3DataNode"
    },
    {
      "failureAndRerunMode": "CASCADE",
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "resourceRole": "DataPipelineDefaultResourceRole",
      "role": "DataPipelineDefaultRole",
      "pipelineLogUri": "s3://dynamodb-import-logs/",
      "scheduleType": "cron",
      "name": "Default",
      "id": "Default"
    },
    {
      "output": {
        "ref": "DataNodeId_h3tjF"
      },
      "input": {
        "ref": "DataNodeId_KlnIq"
      },
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "name": "CSVImportActivity",
      "hiveScript": "INSERT OVERWRITE TABLE ${output1} SELECT * FROM ${input1};",
      "id": "ActivityId_lGBRA",
      "runsOn": {
        "ref": "EmrClusterId_gxpFR"
      },
      "type": "HiveActivity"
    },
    {
      "occurrences": "1",
      "period": "1 Day",
      "name": "RunOnce",
      "id": "DefaultSchedule",
      "type": "Schedule",
      "startAt": "FIRST_ACTIVATION_DATE_TIME"
    },
    {
      "schedule": {
        "ref": "DefaultSchedule"
      },
      "name": "TestCSVImportCluster",
      "coreInstanceType": "m1.medium",
      "coreInstanceCount": "1",
      "masterInstanceType": "m1.medium",
      "amiVersion": "3.8.0",
      "id": "EmrClusterId_gxpFR",
      "region": "eu-west-1",
      "type": "EmrCluster",
      "terminateAfter": "10 Hours"
    },
    {
      "name": "OutputDynamoDBDataFormat",
      "column": [
        "col_a string",
        "col_b string",
        "col_c string",
        "col_d string",
        "col_e string",
        "col_f string",
        "col_g string",
        "col_h bigint",
        "col_i bigint"
      ],
      "id": "DynamoDBDataFormatId_suGD7",
      "type": "DynamoDBDataFormat"
    }
  ],
  "parameters": []
}