E-MapReduce:Sync EMR Kafka data to Hive

Prerequisites

Before you begin, ensure that you have:

• An EMR data lake cluster with Flume selected as an optional service. For more information, see Create a cluster

• An EMR Dataflow cluster with Kafka selected as an optional service. For more information, see Create a cluster

Step 1: Create a Hive table

Log on to the data lake cluster in SSH mode. For more information, see Log on to a cluster.

Run the following statement to create a table named flume_test:

create table flume_test (id int, content string)
clustered by (id) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true');

Step 2: Configure Flume

1. Go to the Configure tab of the Flume service.

   1. Log on to the EMR console. In the left-side navigation pane, click EMR on ECS.

   2. In the top navigation bar, select the region where your cluster resides and select a resource group.

   3. On the EMR on ECS page, find the data lake cluster and click Services in the Actions column.

   4. On the Services tab, click Configure in the Flume service section.

2. On the Configure tab, click the flume-conf.properties subtab. By default, the global configuration applies to all nodes. To configure individual nodes, select Independent Node Configuration from the drop-down list.

3. Add the following configuration to the flume-conf.properties value:

   Parameter	Default	Description
   default-agent.sources.source1.kafka.bootstrap.servers	—	Hostnames and port numbers of brokers in the Kafka cluster. Format: host1:port1,host2:port2,...
   default-agent.sinks.k1.hive.metastore	—	URI of the Hive metastore. Format: thrift://emr-header-1.cluster-xxx:9083. Run hostname on the emr-header-1 node to get the hostname.
   default-agent.channels.c1.capacity	100	Maximum number of events stored in the channel. Adjust based on your throughput requirements.
   default-agent.channels.c1.transactionCapacity	100	Maximum number of events each transaction receives from the source or delivers to the sink. Adjust based on your throughput requirements.

   default-agent.sources = source1
   default-agent.sinks = k1
   default-agent.channels = c1
   
   default-agent.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
   default-agent.sources.source1.channels = c1
   default-agent.sources.source1.kafka.bootstrap.servers = <kafka-host1:port1,kafka-host2:port2...>
   default-agent.sources.source1.kafka.topics = flume-test
   default-agent.sources.source1.kafka.consumer.group.id = flume-test-group
   
   # Describe the sink
   default-agent.sinks.k1.type = hive
   default-agent.sinks.k1.hive.metastore = thrift://xxxx:9083
   default-agent.sinks.k1.hive.database = default
   default-agent.sinks.k1.hive.table = flume_test
   default-agent.sinks.k1.serializer = DELIMITED
   default-agent.sinks.k1.serializer.delimiter = ","
   default-agent.sinks.k1.serializer.serdeSeparator = ','
   default-agent.sinks.k1.serializer.fieldnames =id,content
   
   default-agent.channels.c1.type = memory
   default-agent.channels.c1.capacity = 100
   default-agent.channels.c1.transactionCapacity = 100
   
   default-agent.sources.source1.channels = c1
   default-agent.sinks.k1.channel = c1

   The following table describes the key parameters. Required parameters are in bold.

4. Save the configuration.

   1. Click Save in the lower-left corner.

   2. In the dialog box that appears, enter an execution reason and click Save.

Step 3: Start the Flume agent

1. On the Status tab of the Flume service, find the FlumeAgent component.

2. In the Actions column, choose More > Restart.

3. In the dialog box that appears, enter an execution reason and click OK.

4. In the Confirm message, click OK.

Step 4: Test data synchronization

1. Use Secure Shell (SSH) to log on to the Dataflow cluster. For more information, see Log on to a cluster.

2. Create a Kafka topic named flume-test:

   kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper master-1-1:2181/emr-kafka --topic flume-test --create

3. Start a Kafka producer and send test data:

   kafka-console-producer.sh --topic flume-test --broker-list master-1-1:9092

   For example, type abc and press Enter.

4. Log on to the data lake cluster in SSH mode, and run the following Hive parameters to enable transaction support:

   set hive.support.concurrency=true;
   set hive.exec.dynamic.partition.mode=nonstrict;
   set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;

5. Query the flume_test table to verify that the data was written:

   select * from flume_test;

   The expected output is:

   OK
   1    a