Sahil/Komal | 4 | Consume messages for France from Kafka and write to…

… HDFS france data folder

CitibikeApiProducer/src/main/resources/application-station-france.properties

@@ -0,0 +1,4 @@
+producer.url=https://api.citybik.es/v2/networks/ford-gobike
+producer.topic=station_data_france
+producer.producerId=producer_station-france
+producer.cron=1/10 * * * * *

docker/docker-compose.yml

@@ -131,6 +131,23 @@ services:
           - streaming-data-internal
         restart: on-failure
 
+    raw-station-data-france-saver:
+      build:
+        context: ../RawDataSaver
+        dockerfile: Dockerfile-station-data-France
+      depends_on:
+        - kafka
+        - station-information-producer
+        - station-status-producer
+#        - station-san-francisco-producer TODO: Add a Producer as a part of #3
+        - hadoop-seed
+        - zookeeper-seed
+      volumes:
+        - ~/.ivy2:/root/.ivy2:rw
+      networks:
+        - streaming-data-internal
+      restart: on-failure
+
     station-consumer:
         build:
             context: ../StationConsumer

sbin/deploy.sh

@@ -90,6 +90,7 @@ echo "====Runing Producers Killed===="
 echo "====Deploy Producers===="
 nohup java -jar /tmp/tw-citibike-apis-producer0.1.0.jar --spring.profiles.active=\${station_information} --kafka.brokers=kafka.${TRAINING_COHORT}.training:9092 1>/tmp/\${station_information}.log 2>/tmp/\${station_information}.error.log &
 nohup java -jar /tmp/tw-citibike-apis-producer0.1.0.jar --spring.profiles.active=\${station_san_francisco} --producer.topic=station_data_sf --kafka.brokers=kafka.${TRAINING_COHORT}.training:9092 1>/tmp/\${station_san_francisco}.log 2>/tmp/\${station_san_francisco}.error.log &
+nohup java -jar /tmp/tw-citibike-apis-producer0.1.0.jar --spring.profiles.active=\${station_france} --producer.topic=station_data_france --kafka.brokers=kafka.${TRAINING_COHORT}.training:9092 1>/tmp/\${station_france}.log 2>/tmp/\${station_france}.error.log &
 nohup java -jar /tmp/tw-citibike-apis-producer0.1.0.jar --spring.profiles.active=\${station_status} --kafka.brokers=kafka.${TRAINING_COHORT}.training:9092 1>/tmp/\${station_status}.log 2>/tmp/\${station_status}.error.log &
 
 echo "====Producers Deployed===="

setup.sh

@@ -47,6 +47,11 @@ $zk_command create /tw/stationDataSF/topic station_data_sf
 $zk_command create /tw/stationDataSF/checkpointLocation /tmp/tw/rawData/stationDataSF/checkpoints
 $zk_command create /tw/stationDataSF/dataLocation /tmp/tw/rawData/stationDataSF/data
 
+$zk_command create /tw/stationDataFrance ''
+$zk_command create /tw/stationDataFrance/kafkaBrokers 127.0.0.1:9092
+$zk_command create /tw/stationDataFrance/topic station_data_france
+$zk_command create /tw/stationDataFrance/checkpointLocation /tmp/tw/rawData/stationDataFrance/checkpoints
+$zk_command create /tw/stationDataFrance/dataLocation /tmp/tw/rawData/stationDataFrance/data
 
 $zk_command create /tw/output ''
 $zk_command create /tw/output/checkpointLocation /tmp/tw/stationMart/checkpoints

zookeeper/seed.sh

@@ -25,6 +25,12 @@ $zk_command create /tw/stationDataSF/topic station_data_sf
 $zk_command create /tw/stationDataSF/checkpointLocation hdfs://$hdfs_server/tw/rawData/stationDataSF/checkpoints
 $zk_command create /tw/stationDataSF/dataLocation hdfs://$hdfs_server/tw/rawData/stationDataSF/data
 
+$zk_command create /tw/stationDataFrance ''
+$zk_command create /tw/stationDataFrance/kafkaBrokers $kafka_server
+$zk_command create /tw/stationDataFrance/topic station_data_france
+$zk_command create /tw/stationDataFrance/checkpointLocation hdfs://$hdfs_server/tw/rawData/stationDataFrance/checkpoints
+$zk_command create /tw/stationDataFrance/dataLocation hdfs://$hdfs_server/tw/rawData/stationDataFrance/data
+
 $zk_command create /tw/output ''
 $zk_command create /tw/output/checkpointLocation hdfs://$hdfs_server/tw/stationMart/checkpoints
 $zk_command create /tw/output/dataLocation hdfs://$hdfs_server/tw/stationMart/data