Simple Java Program for publishing Syslog Events
In the Using Syslog as source in Flume i blogged about how to configure flume to listen for Syslog event on particular UDP port. I wanted to test that configuration so i built this simple java program that can publish Syslog event on given host and port no.
You can download the source code for this project from GitHub
This program takes 3 arguments first is hostname for the syslog server, second is the port on which the server is listening and third is the actual message that you want to send.
Using Syslog as source in Flume
I wanted to figure out how to use Flume for receiving Syslog message. So i tried 2 different configurations one is using Syslog server on TCP port and other on UDP port.
This is the flume configuration for listening on UDP port
Copy the flumesyslogudp.properties file in the conf directory of your flume server and use following command to start flume server
bin/flume-ng agent --conf conf --conf-file conf/flumesyslogudp.properties --name agent1
-Dflume.root.logger=DEBUG,console
Or you can configure flume to listen on TCP port. Only difference is the source type is syslogtcp instead of syslogudp
bin/flume-ng agent --conf conf --conf-file conf/flumesyslogtcp.properties --name agent1
-Dflume.root.logger=DEBUG,console
Read configuration from .properties file in
This is small utility function in Scala, that takes fully qualified path of the properties file, and converts it into Map and returns. I use it for taking path of the properties file in my standalone Scala program and load it into Map
def getConfig(filePath: String)= {
Source.fromFile(filePath).getLines().filter(line => line.contains("=")).map{ line =>
println(line)
val tokens = line.split("=")
( tokens(0) -> tokens(1))
}.toMap
}
Once you have this method you can call it like this getConfig(filePath)
Flume Hello World tutotiral
I am using flume for some time now and really like it. This is simple HelloWorld tutorial that i thought would be helpful if you want to get started with Flume. This tutorial will walk you through steps for setting up Flume that listens to messages on port 44444, once it gets message it just prints it out on console, Follow these steps
- First create sampleflume.properties file on your machine like this
Your flume configuration file must have at least 3 elements a source, channel and sink# example.conf: A single-node Flume configuration # Name the components on this agent agent1.sources = netcat1 agent1.sinks = logger1 agent1.channels = memory1 # Describe/configure the source agent1.sources.netcat1.type = netcat agent1.sources.netcat1.bind = localhost agent1.sources.netcat1.port = 44444 # Describe the sink agent1.sinks.logger1.type = logger # Use a channel which buffers events in memory agent1.channels.memory1.type = memory agent1.channels.memory1.capacity = 1000 agent1.channels.memory1.transactionCapacity = 100 # Bind the source and sink to the channel agent1.sources.netcat1.channels = memory1 agent1.sinks.logger1.channel = memory1- netcat1: netcat1 source defines how flume is listening to messages. In this case type of netcat means it will listen on port that you can connect to using either netcat or telnet
- memory: memory channel defines how flume stores messages that it has received before they are consumed by sink. In this case i am saying keep the messages in memory
- logger1: Logger sink is for testing, it just prints the messages on console
-
Once your configuration file is ready you can start a flume agent by executing following command
YOu will see flume printing messages on the console while it is starting like thisflume-ng agent --conf conf --conf-file sampleflume.properties --name agent1 -Dflume.root.logger=DEBUG,console
-
Once server is started you can connect to it using nc or telnet and send messages to it like this. Whatever messages you send will be printed to console
- Once you send messages using nc command look at the server console and you should see the messages that you sent
Configuring Flume to write avro events into HDFS
Recently i wanted to figure out how to configure Flume so that it can listen for Avro Events and whenever it gets event it should dump it in the HDFS. In order to do that i built this simple Flume configuration
# example.conf: A single-node Flume configuration
# Name the components on this agent
agent1.sources = avro
agent1.sinks = logger1
agent1.channels = memory1
# Describe/configure the source
agent1.sources.avro.type = avro
agent1.sources.avro.bind = localhost
agent1.sources.avro.port = 41414
agent1.sources.avro.selector.type = replicating
agent1.sources.avro.channels = memory1
# Describe the sink
agent1.sinks.hdfs1.type = hdfs
agent1.sinks.hdfs1.hdfs.path=/tmp/flume/events
agent1.sinks.hdfs1.hdfs.rollInterval=60
#The number of events to be written into a file before it is rolled.
agent1.sinks.hdfs1.hdfs.rollSize=0
agent1.sinks.hdfs1.hdfs.batchSize=100
agent1.sinks.hdfs1.hdfs.serializer=org.apache.flume.sink.hdfs.AvroEventSerializer$Builder
agent1.sinks.hdfs1.hdfs.fileType = DataStream
agent1.sinks = hdfs1
agent1.sinks.hdfs1.channel = memory1
# Use a channel which buffers events in memory
agent1.channels.memory1.type = memory
agent1.channels.memory1.capacity = 1000
agent1.channels.memory1.transactionCapacity = 100
In this i have Avro source listening on local machine at port 41414, once it gets event it writes that in HDFS in /tmp/flume/events directory
Once this file is saved in local machine as hellohdfsavro.conf i can start the flume agent using following command
flume-ng agent --conf conf --conf-file hellohdfsavro.conf --name agent1 -Dflume.root.logger=DEBUG,console
Configure Flume to use IBM MQ as JMS Source
Recently i had a requirement in which i wanted to figure out how to read XML documents stored as message in IBM MQ and post them into Hadoop. I decided to use Apache Flume + Flume JMS Source + Flume HDFS Sink for this. I had to use following steps for this setup. Please note that i am not WebSphere MQ expert so there might be a better/easier way to achieve this.
- First i had to install WebSphere MQ Client on my windows machine
- Next i did create a simple jms.config like this in c:\temp folder of my windows box
INITIAL_CONTEXT_FACTORY=com.sun.jndi.fscontext.RefFSContextFactory PROVIDER_URL=file:/C:/temp/jmsbinding - Next step is to run JMSAdmin.bat c:\temp\jms.config, it opens up a console like this, type following command in it and change it to use the right configuration that you need
Once you execute this command it will generate .bindings file in C:/temp/jmsbinding (Folder that is configured as value of PROVIDER_URL)DEF CF(myConnectionFactory) QMGR(myQueueManager) HOSTNAME(myHostName) PORT(1426) CHANNEL(myChannelName) TRANSPORT(CLIENT) - Next step for me was to copy the C:/temp/jmsbinding/.bindings folder to
/etc/flume/conffolder in my linux box which has Flume running on it. -
In addition to bindings file i also need the MQ client jar files. I started by copying jms.jar from
C:\Program Files (x86)\IBM\WebSphere MQ\java\libto/usr/hdp/current/flume-server/lib/folder in my Hadoop installation, but i kept getting ClassNotFoundException and to deal with that i copied more and more jars from my MQ Client into Flumejms.jar fscontext.jar jndi.jar providerutil.jar com.ibm.mq.jar com.ibm.mqjms.jar com.ibm.mq.pcf.jar connector.jar dhbcore.jar com.ibm.mq.jmqi.jar com.ibm.mq.headers.jar -
Once the Flume MQ setup was in place, last step was to create Flume Configuration that points to your bindings file and also points to your MQ server like this
# Flume agent config #st the sources, channels, and sinks for the agent ggflume.sources = jms ggflume.channels = memory ggflume.sinks = hadoop ggflume.sources.jms.channels=memory ggflume.sinks.hadoop.channel=memory ggflume.sources.jms.type = jms ggflume.sources.jms.providerURL = file:///etc/flume/conf ggflume.sources.jms.initialContextFactory = com.sun.jndi.fscontext.RefFSContextFactory ggflume.sources.jms.destinationType=QUEUE ggflume.sources.jms.destinationName=<channelName> ggflume.sources.jms.connectionFactory=myConnectionFactory ggflume.sources.jms.batchSize=1 ggflume.channels.memory.type = memory ggflume.channels.memory.capacity = 1000 ggflume.channels.memory.transactionCapacity = 100 ggflume.sinks.hadoop.type=hdfs ggflume.sinks.hadoop.hdfs.path=/data/mq/xml ggflume.sinks.hadoop.hdfs.filePrefix=sample -
Now start flume server by executing following flume command
flume-ng agent --conf conf --conf-file mqflume.conf --name ggflume -Dflume.root.logger=DEBUG,console
Moving data from Avro to ORC files
In the Importing data from Sqoop into Hive External Table with Avro encoding i blogged about how to sqoop data from RDBMS into Hive. But i wanted to take it to next step by moving the data downloaded to ORC table. I followed these steps to achieve that
- First thing is to find out the schema of the table in Avro and you can get that by executing following statement in hive
You will get output that looks something like this, it contains schema of the tableshow create table CUSTOMER;CREATE EXTERNAL TABLE `CUSTOMER`( `contactid` int COMMENT 'from deserializer', `firstname` string COMMENT 'from deserializer', `lastname` string COMMENT 'from deserializer', `email` string COMMENT 'from deserializer') ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe' STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat' LOCATION 'hdfs://sandbox.hortonworks.com:8020/tmp/customer/data' TBLPROPERTIES ( 'avro.schema.url'='hdfs:///tmp/customer/schema/customer.avsc', 'transient_lastDdlTime'='1431719666') -
Copy the schema from last step and remove the part about format and table properties and replace it with part that highlighted in this code, execute this in Hive to create customer table in ORC format
CREATE EXTERNAL TABLE `CUSTOMER_ORC`( `contactid` int , `firstname` string , `lastname` string , `email` string ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\001' LOCATION 'hdfs://sandbox.hortonworks.com:8020/tmp/customer/data_orc' STORED AS ORC tblproperties ("orc.compress"="SNAPPY","orc.row.index.stride"="20000"); -
Last step is to copy data from avro table to ORC table, you can achieve that by using following command
insert into table CUSTOMER_ORC select * from customer;
Importing data from Sqoop into Hive External Table with Avro encoding
I wanted to figure out how to import content of RDBMS table into Hive with Avro encoding, during this process i wanted to use external hive tables so that i have complete control over the location of files.
Note: I have a different/easier method for doing this in Importing data from Sqoop into Hive External Table with Avro encoding updated
First i did create following table in the mysql database which is on the same machine as that of my HortonWorks Sandbox
Note: I have a different/easier method for doing this in Importing data from Sqoop into Hive External Table with Avro encoding updated
First i did create following table in the mysql database which is on the same machine as that of my HortonWorks Sandbox
- First create CUSTOMER table like this in mysql
CREATE TABLE CUSTOMER ( contactid INTEGER NOT NULL , firstname VARCHAR(50), lastname VARCHAR(50), email varchar(50) ); - After creating table add couple of records in it by executing following insert statement
insert into customer values(1,'Sachin','Tendulark','sachin@gmail.com'); -
Next step is to run sqoop query that downloads records of the table into HDFS at /tmp/customer/sample. In real world you might want to download only first 10 records or so into Hive, because you need few sample records just to create avro schema
sqoop import --connect jdbc:mysql://localhost/test --table CUSTOMER --username sqoop1 --password sqoop -m 1 --create-hive-table --hive-table CONTACT --as-avrodatafile --target-dir /tmp/customer/sample - Running sqoop command it will dump records in HDFS, so first download the avro file generated by sqoop
hdfs dfs -get /tmp/customer/sample/part-m-00000.avro - Use the avro-tools-*.jar, to read schema of the file generated by sqoop. by executing following command
This is how the customer.avsc file looks like in my casejava -jar avro-tools-1.7.5.jar getschema part-m-00000.avro > customer.avsc{ "type" : "record", "name" : "CUSTOMER", "doc" : "Sqoop import of CUSTOMER", "fields" : [ { "name" : "contactid", "type" : [ "int", "null" ], "columnName" : "contactid", "sqlType" : "4" }, { "name" : "firstname", "type" : [ "string", "null" ], "columnName" : "firstname", "sqlType" : "12" }, { "name" : "lastname", "type" : [ "string", "null" ], "columnName" : "lastname", "sqlType" : "12" }, { "name" : "email", "type" : [ "string", "null" ], "columnName" : "email", "sqlType" : "12" } ], "tableName" : "CUSTOMER" } -
Next step is to upload the avro schema file that you created in the last step back to HDFS, in my case i had HDFS folder called
/tmp/customer/schemaand i uploaded the avro schema file in ithdfs dfs -put customer.avsc /tmp/customer/schema/ - Now go to hive and execute the following command to define External Customer Hive table with avro schema defined in last step
CREATE EXTERNAL TABLE CUSTOMER ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe' STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat' LOCATION '/tmp/customer/data' TBLPROPERTIES ('avro.schema.url'='hdfs:///tmp/customer/schema/customer.avsc'); -
Last step is to run sqoop again but this time with all the data in the external directory that Customer hive table is pointing to.
sqoop import --connect jdbc:mysql://localhost/test --table CUSTOMER --username sqoop1 --password sqoop -m 1 --as-avrodatafile --target-dir /tmp/customer/data --compression-codec snappy
Running oozie job on Hortonworks Sandbox
In the Enabling Oozie console on Cloudera VM 4.4.0 and executing examples i blogged about how to run oozie job in Cloudera Sandbox. It seems this process is little bit easier in HortonWorks 2.2 sandbox.
So first i had brand new HDP 2.2 image and i tried running oozie example on it by executing
oozie job -oozie http://localhost:11000/oozie -config examples/apps/map-reduce/job.properties -run
But when i tried running it i got following error
Error: E0501 : E0501: Could not perform authorization operation, Call From sandbox.hortonworks.com/10.0.2.15 to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
So i looked into /var/log/oozie/oozie.log and i saw following error
2015-05-01 20:34:39,195 WARN V1JobsServlet:546 - SERVER[sandbox.hortonworks.com] USER[root] GROUP[-] TOKEN[-] APP[-] JOB[-] ACTION[-] URL[POST http://sandbox.hortonworks.com:11000/oozie/v2/jobs?action=start] error[E0501], E0501: Could not perform authorization operation, Call From sandbox.hortonworks.com/10.0.2.15 to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
org.apache.oozie.servlet.XServletException: E0501: Could not perform authorization operation, Call From sandbox.hortonworks.com/10.0.2.15 to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
at org.apache.oozie.servlet.BaseJobServlet.checkAuthorizationForApp(BaseJobServlet.java:240)
at org.apache.oozie.servlet.BaseJobsServlet.doPost(BaseJobsServlet.java:96)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:727)
at org.apache.oozie.servlet.JsonRestServlet.service(JsonRestServlet.java:287)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:820)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:290)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
at org.apache.oozie.servlet.AuthFilter$2.doFilter(AuthFilter.java:143)
at org.apache.hadoop.security.authentication.server.AuthenticationFilter.doFilter(AuthenticationFilter.java:572)
at org.apache.hadoop.security.authentication.server.AuthenticationFilter.doFilter(AuthenticationFilter.java:542)
at org.apache.oozie.servlet.AuthFilter.doFilter(AuthFilter.java:148)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
at org.apache.oozie.servlet.HostnameFilter.doFilter(HostnameFilter.java:84)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:235)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:233)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:191)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:127)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:103)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:109)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:293)
at org.apache.coyote.http11.Http11Processor.process(Http11Processor.java:861)
at org.apache.coyote.http11.Http11Protocol$Http11ConnectionHandler.process(Http11Protocol.java:606)
at org.apache.tomcat.util.net.JIoEndpoint$Worker.run(JIoEndpoint.java:489)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.oozie.service.AuthorizationException: E0501: Could not perform authorization operation, Call From sandbox.hortonworks.com/10.0.2.15 to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
at org.apache.oozie.service.AuthorizationService.authorizeForApp(AuthorizationService.java:399)
at org.apache.oozie.servlet.BaseJobServlet.checkAuthorizationForApp(BaseJobServlet.java:229)
... 25 more
Caused by: java.net.ConnectException: Call From sandbox.hortonworks.com/10.0.2.15 to localhost:8020 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
at org.apache.hadoop.net.NetUtils.wrapWithMessage(NetUtils.java:791)
at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:731)
at org.apache.hadoop.ipc.Client.call(Client.java:1472)
at org.apache.hadoop.ipc.Client.call(Client.java:1399)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232)
at com.sun.proxy.$Proxy29.getFileInfo(Unknown Source)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.getFileInfo(ClientNamenodeProtocolTranslatorPB.java:752)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
at com.sun.proxy.$Proxy30.getFileInfo(Unknown Source)
at org.apache.hadoop.hdfs.DFSClient.getFileInfo(DFSClient.java:1988)
at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:1118)
at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:1114)
at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1114)
at org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:1400)
at org.apache.oozie.service.AuthorizationService.authorizeForApp(AuthorizationService.java:371)
... 26 more
Caused by: java.net.ConnectException: Connection refused
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739)
at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWithTimeout.java:206)
at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:530)
at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:494)
at org.apache.hadoop.ipc.Client$Connection.setupConnection(Client.java:607)
at org.apache.hadoop.ipc.Client$Connection.setupIOstreams(Client.java:705)
at org.apache.hadoop.ipc.Client$Connection.access$2800(Client.java:368)
at org.apache.hadoop.ipc.Client.getConnection(Client.java:1521)
at org.apache.hadoop.ipc.Client.call(Client.java:1438)
... 44 more
In order to solve these issues i had to make changes in examples/apps/map-reduce/job.properties, to replace localhost with sandbox.hortonworks.com
nameNode=hdfs://sandbox.hortonworks.com:8020
jobTracker=sandbox.hortonworks.com:8032
queueName=default
examplesRoot=examples
oozie.wf.application.path=${nameNode}/user/${user.name}/${examplesRoot}/apps/map-reduce
outputDir=map-reduce
Exporting data from Hive table to RDBMS
In the Importing data from RDBMS into Hadoop using sqoop i blogged about how to import data from RDBMS to Hive, but now i wanted to figure out how to export data from Hive back to RDBMS, Sqoop has export feature that allows you to export data from Hadoop directory(CSV files in a directory) to RDBMS,
I wanted to try exporting data from sqoop so first i created a simple contact_hive table and populated some data in it, then i used sqoop to export the content of contact_hive table into contact table in MySQL, i followed these steps, if you already have a hive table populated then you can skip first 5 steps and go to step 6.
-
Create contacthive.csv file which has simple data with 4 columns separated by comma
1,MahendraSingh,Dhoni,mahendra@bcci.com 2,Virat,Kohali,virat@bcci.com 5,Sachin,Tendulkar,sachin@bcci.com -
Upload the contacthive.csv that you created in last step in HDFS at /tmp folder using following command
hdfs dfs -put contacthive.csv /tmp -
Define a contact_hive table that will have 4 columns, contactId, firstName, lastName and email, execute this command in hive console
CREATE TABLE contact_hive(contactId Int, firstName String, lastName String, email String) row format delimited fields terminated by "," stored as textfile; -
In this step populate the contact_hive table that you created in the last step with the data from contacthive.csv file created in step 1. Execute this command in Hive console to populate contact_hive table
LOAD DATA INPATH "/tmp/contacthive.csv" OVERWRITE INTO TABLE contact_hive; -
Since i am using Hive managed table, it will move the contacthive.csv file to Hive managed directory in case of Hortonworks that directory is
/apps/hive/warehouse, You can verify that by executing following command on HDFShdfs dfs -ls /apps/hive/warehouse/contact_hive - Before you export data into RDBMS, you will have to create the table in mysql, use following command to create the CONTACT table in mysql.
CREATE TABLE CUSTOMER ( contactid INTEGER NOT NULL , firstname VARCHAR(50), lastname VARCHAR(50), email varchar(50) ); -
Now last step is to execute sqoop export command that exports data from hive/hdfs directory to database
sqoop export --connect jdbc:mysql://localhost/test --table CONTACT --export-dir /apps/hive/warehouse/contact_hive
Subscribe to:
Posts (Atom)


