1. Hadoop and NoSQL: ScalableBack-end Clusters Orchestrationin Real-world SystemsCloudCon 2012, Dalian, ChinaRuo Ando NICT National Institute of Information and Communications Technology, Tokyo Japan

2. Agenda: Scalable Back-end Clusters Orchestration for real-world systems (large scale network monitoring)Hadoop and NoSQL: Scalable Back-end Clusters Orchestration in Real-world SystemsHadoop and NoSQl are usually used together. Partly because Key-Value data format (such as JSON)is suitable for exchanging data between MongoDB and HDFS. These technologies is deployed networkmonitoring system and large scale Testbed in National research institute in Japan.What is Orchestration is for? large scale network monitoringWith rapid enragement of botNet and file sharing networks, network traffic monitoring logs has becomebig data. Todays Large scale network monitoring needs scalable clusters for traffic logging and dataprocessing.Back ground Internet traffic explosionSome statistics are shown about mobile phone traffic and "gigabyte club"Real world systems large scale DHT network crawlingTo test performance of our system, we have crawling DHT (BitTorrent) network. Our system have obtainedinformation of over 10,000,000 nodes in 24 hours. Besides, ranking of countries about popularity of DHTnetwork is generated by our HDFS.Architecture overviewWe use everything available for constructing high-speed and scalable clusters (hypervisor, NoSQL, HDFS,Scala, etc..)Map Reduce and Traffic logsFor aggregating and sorting traffic logs, we have programmed two stage Map Reduce.Results and demosconclusion 3. NICT: National Institute of Information andSolar observatoryCommunications Technology, Tokyo Japan Large scale TestBedsLarge scale network emulation foranalyzing cyber incidents (DDOS, BotNet) We have over 140,000 passivemonitor in Darknet for analyzing botNetDarknet monitoring for malware analysis 4. StarBed:A Large Scale Network Experiment Environment in NICT Developers along desire to evaluate their new technologies in realistic situations. The developers for the Internet are not excepted. The general experimental issues for Internet technologies are efficiency and scalability. StarBED enables to evaluate such factors in realistic situations. Actual computers and network equipments are required if we want to evaluate software for the real Internet. In StarBED there are many actual computers, and switches which connect these computers. We reproduce close to reality situations with actual equipments that are used on Internet. If developers want to evaluate their real implementation, they have to use actual equipments. group # of experiment networks F 168 0 0 4 SATA 2006 H 240 0 0 2 SATA 2009 I 192 0 0 4 SATA 2011 J 96 0 0 4 SATA 2011 There are about 1000 servers. Other 500 StarBed collaborates with other testbed project of total 960DETER, PlanetLab in US.Group I,J,K,L Model Cisco UCS C200 M2 CPU Intel 6-Core Xeon X5670 x 2Memory 48.0GB Disk SATA 500GB x 2 Network (on-board) double GigabitEthernet 5. Real world systems: monitoring Bittorrent network - handling massive DHT crawling Invisibility (thus unstoppable)encourages illegal adoption of DHT networkBit Torrent traffic rate of all internetestimatesIn 2010 Oct, A New York judge ordered LimeWire 55% - CableLabsto shutdown its file-sharing software.About an half of upstream traffic of CATV. US federal court judge issued that Limewires 35% - CacheLogicservice is used as one of the software forLIVEWIRE - File-sharing network thrives beneathinfringement of copyright contents.the Radar Later soon, the new version of Limewire called 60% - documents in www.sans.eduLPE (Limewire Pirate Edition) has been released It is estimated that more than 60% of the traffic onas resurrection by anonymous creators. the internet is peer-to-peer. 6. Parser and translator isArchitecture Overview parallelized by Scala. Virtual machines and Data nodes is applicable for scaling out. 7. Rapid crawling: 24 hours to reach NoSQL has stored10000000 peers !10,000,000 peersnode1200000010000000 8000000 6000000 4000000 2000000 0 0 1 2 3 4 5 6 7 8 9101112 13141516 17 18 19 20 21 22 23 24 25 26hourdiff 1000000100000 10000 0 1 2 3 4 5 6 7 8 9 10111213 14151617 18 19 20 21 22 23 24 25 26 8. Demo: visualizing propagation of DHT crawlingWe have crawledmore than 10,000,000 Peers in DHT nework In 24 hoursSQL (MySQLor Postgres) Cannot handle 4,000,000peers in3 hours ! 9. DHT crawler and Map Reduce For huge scale of DHT network, we cannotWithout HDFS, it takes 7 days forrun too many crawlers. processing data of 1 day. RANKCountry # of nodes RegionDomain1Russia 1,488,056 Russia RU2United states1,177,766 North AmericaUS3China815,934 East AsiaCN4UK 414,282 West EuropeGB5Canada 408,592 North AmericaCA6Ukraine399,054 East EuropeUA7France 394,005 West EuropeFR8India309,008 South Asia IN9Taiwan 296,856 East AsiaTWDHT network10 Brazil 271,417 South AmericaBR11 Japan262,678 East AsiaJP12 Romania233,536 East EuropeRO13 Bulgaria 226,885 East EuropeBG14 South Korea217,409 East AsiaKR15 Australia216,250 OceaniaAU16 Poland 184,087 East EuropePL17 Sweden 183,465 North Europe SE18 Thailand 183,008 South East AsiaTH19 Italy177,932 West EuropeIT20 Spain172,969 West EuropeES ReduceDHT Crawler DHT Crawler DHT Crawler Shuffle Scale out ! Map Map MapKey value store =node ID =data (address, port, etc) Dump Data Map job should be increased corresponding to the number of DHT crawler. 10. Scaling DHT crawlers out!FIND_NODE : used to obtain thecontact information of ID.Response should be a key nodes or the compact nodeinfo for the target node or the K (8) in its routing table.arguments: {"id" : "","target" : ""}response: {"id" : "","nodes" : ""}DHT network The response should be a key nodes of or the compact node info for the target node or the K (8) in its routing table.DHT CrawlerDHT Crawler DHT Crawler Info of key nodes and K(8) should be Hypervisorrandomly distributed. So we can obtain 8^N peers in worst case. 11. Rapid propagation ofDHT gossip protocol N^Mnode 12000000 100000008000000600000040000002000000 0 0 1 2 3 4 5 6 7 8 9101112 1314151617181920212223242526diff 1000000 Applying100000gossip protocol, 10000 0 1 2 3 4 5 6 7 8 9 10111213 14151617181920212223242526DHT has N^M (N=5-8)After 5 hours, increasing) propagation become stablespeed. In first 4 hours, we can obtain more than 4000000 peers! 12. Visualization & ranking 77.221.39.201,6881,2011/9/25 23:57:43,1 87.97.210.128,62845,2011/9/25 23:56:32,1 188.40.33.212,6881,2011/9/25 23:33:58,1 188.232.9.21,49924,2011/9/25 23:37:02,1 Traffic logsis parsedInto XML Location info is (Keyhole IP addressretrieved by GeoIPTime Markup from each IP address Language)Location InfoDomain name (country, city, latlng)KML movieStrings are tokenized Figure and aggregated rankingby HDFS 13. Two-Stage Map Reduce: count and sorting Frequency countSorting accordingfor each word to Reduce1 MapReduce1MapInputMapReduceOutput Reduce2 Map MapMapReduce is the algorithm suitable for coping with Big data.Ranking (sorting)map(key1,value) -> listNeed second stagereduce(key2, list) -> listof Map phase.MapReduce: Simplified Data Processing on Large ClustersJeffrey Dean and Sanjay GhemawatOSDI04: Sixth Symposium on Operating System Design and Implementation,San Francisco, CA, December, 2004. 14. Map Phase *.0.194.107,h116-0-194-107.catv02.itscom.jp *.28.27.107,c-76-28-27-107.hsd1.ct.comcast.net *.40.239.181,c-68-40-239-181.hsd1.mi.comcast.net *.253.44.184,pool-96-253-44-184.prvdri.fios.verizon.net *.27.170.168,cpc11-stok15-2-0-cust167.1-4.cable.virginmedia.com *.22.23.81,cpc2-stkn10-0-0-cust848.11-2.cable.virginmedia.com*.0.194.107 hdsl1 comcasthdsl1 comcastverizon virginmedia 11 11 1 11Log string is divided into words and assigned 1.key-value {word, 1}Map job iseasier to increaseIn Map phase, each line is tokenized for a word, and each wordthen Reduce job. is assigned 1. 15. Reduce Phase*.0.194.107hdsl1comcasthdsl1 comcastverizon virginmedia 1 111 1 11 hdsl1comcast verizon 1 1 1 1 1Reduce job is applied for counting frequency of each word.Reduce: count up 1 for each word.Key-value {hdsl, 2} / Key-value {comcast, 2} / Key-value {verizon, 1} 16. Sorting and ranking*.0.194.107hdsl1comcast hdsl1comcast verizonhdsl1 1 11 11 11 hdsl1comcastverizon1 111 1Sorting and ranking is1 second reduce phase. Words with the frequency is sorted in shuffle.@list1 = reverse sort { (split(/s/,$a))[1] (split(/s/,$b))[1] } @list1; 17. Example: # of nodes Ranking in one day RANKCountry # of nodes RegionDomain1Russia 1,488,056 Russia RU2United states1,177,766 North AmericaUS3China815,934 East AsiaCN4UK 414,282 West EuropeGB5Canada 408,592 North AmericaCA6Ukraine399,054 East EuropeUA7France 394,005 West EuropeFR8India309,008 South Asia IN9Taiwan 296,856 East AsiaTW10 Brazil 271,417 South AmericaBR11 Japan262,678 East AsiaJP12 Romania233,536 East EuropeRO13 Bulgaria 226,885 East EuropeBG14 South Korea217,409 East AsiaKR15 Australia216,250 OceaniaAU16 Poland 184,087 East EuropePL17 Sweden 183,465 North Europe SE18 Thailand 183,008 South East AsiaTH19 Italy177,932 West EuropeIT20 Spain172,969 West EuropeES 18. ALL cities except US N/A 978457 1 Moscow 285097 (RU:1) 2 Beijing 240419 (CN:3) 3 Seoul 180186 (KR) 4 Taipei 161498 (TW:9) 5 Kiev 117392 (RU:1) 6 Saint Petersburg 94560 7 Bucharest 79336These peers has8 Sofia 78445 (BG:13)been connected from9 Central District 65635 (HK)single point in Tokyo in24 hours. Propagation 10 Bangkok 62882 (TH:18)in DHT network is11 Delhi 62563 (IN:8)beyond over12 Tokyo 54531 (JP:11)boarder control. 13 London 53514 (GB:4) 14 Guangzhou52981 (CN:3) 15 Athens 52656 (3680000: 1.4%) 16 Budapest52031Z. N. J. Peterson, M. Gondree, and R. Beverly.A position paper on data sovereignty:The importanceof geolocating data in the cloud.the 3nd USENIX workshop on Hot Topics inCloud Computing, June 2011 19. rank 3 China815,934 East Asia CN name# of peers population Beijing2404191755 Guangzhou 52981 1,004 Shanghai273991921 Jinan 26281 569 Chengdu 188351059 Shenyang18566 776 Tianjin 184601228 Hebei 17414- Wuhan 15239 910 Hangzhou12997 796 Harbin10848 987 Changchun 10411 751 Nanning 10318 648 Beijing is the largest city of which the Qingdao 10257 757 number of peers is about 240000, secondto Moscow. Tokyo 545311318 In china, BT seems to be popular besides Osaka7430886 many domestic file sharing systems. yokohama 6983369 BitComet: a popularTokyo and Guangzhou has almost the samenumber of peers about 50000. client in Asia 20. Demo2: (almost) real time monitoring of peersin Japan In this movie, there are four colors According tothe numberof fileslocated ineach point.In this slide, traffic logis translated into XML Key hole markup LanguageMovie can be generated after a day.Spying the World from your Laptop -- Identifying and Profiling Content Providers andAggregation and translation of 24 hours is Big Downloaders in BitTorrent completed in 16 hours 3rd USENIX Workshop on Large-Scale Exploits and Emergent Threats (LEET10) (2010) 21. Conclusion: Scalable Back-end Clusters Orchestrationfor real-world systems (large scale network monitoring)Hadoop and NoSQL: Scalable Back-end Clusters Orchestration in Real-world SystemsHadoop and NoSQl are usually used together. Partly because Key-Value data format (such as JSON)is suitable for exchanging data between MongoDB and HDFS. These technologies is deployednetwork monitoring system and large scale Testbed in National research institute in Japan.What is Orchestration is for? large scale network monitoringWith rapid enragement of botNet and file sharing networks, network traffic monitoring logs hasbecome big data. Todays Large scale network monitoring needs scalable clusters for trafficlogging and data processing.Back ground Internet traffic explosionSome statistics are shown about mobile phone traffic and "gigabyte club"Real world systems large scale DHT network crawlingTo test performance of our system, we have crawling DHT (BitTorrent) network. Our system haveObtained information of over 10,000,000 nodes in 24 hours. Besides, ranking of countries aboutpopularity of DHT network is generated by our HDFS.Architecture overviewWe use everything available for constructing high-speed and scalable clusters (hypervisor, NoSQL,HDFS, Scala, etc..)Map Reduce and Traffic logsFor aggregating and sorting traffic logs, we have programmed two stage Map Reduce.