devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据框架：（电信诈骗实时查询框架）

作者：Monodyee | 2024-04-05 02:25:19

赞

踩

大数据框架：（电信诈骗实时查询框架）

框架图：

flume:一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。支持在日志系统中定制各类数据发送方，用于收集数据;同时，Flume提供对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力 

kafka:一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决

storm:流计算

mysql:数据库：用于与web的链接。（目前涉及到客户端的都是利用数据库进行交互）

hadoop:分布式大数据框架

hive:数据仓库

spark:基于内存的分布式大数据框架

redis：键值数据库

利用maven 进行依赖项的自行加载，不需要我们手动导包。

看此篇文章部署：eclipse配置maven环境

利用eclipse进行开发。

关于电话诈骗建立的几个类

1.数据库的链接：JDBC

  1. import java.sql.Connection;
  2. import java.sql.DriverManager;
  3. import java.sql.ResultSet;
  4. import java.sql.SQLException;
  5. import java.util.ArrayList;
  6. import java.util.List;
  7.  
  8. import org.apache.commons.dbutils.BasicRowProcessor;
  9. import org.apache.commons.dbutils.QueryRunner;
  10. import org.apache.commons.dbutils.handlers.ArrayListHandler;
  11.  
  12. public final class MyDbUtils {// 拒绝继承
  13. 	private static String className = "com.mysql.jdbc.Driver";
  14. 	private static String url = "jdbc:mysql://192.168.115.130:3306/test?useUnicode=true&characterEncoding=utf-8";
  15. 	private static String user = "root";
  16. 	private static String password = "root";
  17. 	private static QueryRunner queryRunner = new QueryRunner();
  18.  
  19. 	public static final String INSERT_LOG = "INSERT INTO LOG(topdomain,usetime,time) VALUES(?,?,?)";
  20.  
  21. 	// 拒绝new一个实例
  22. 	private MyDbUtils() {
  23. 	};
  24.  
  25. 	static {// 调用该类时既注册驱动
  26. 		try {
  27. 			Class.forName(className);
  28. 		} catch (Exception e) {
  29. 			e.printStackTrace();
  30. 			throw new RuntimeException();
  31. 		}
  32. 	}
  33.  
  34. 	public static void main(String[] args) {
  35. 	}
  36.  
  37. 	public static List<String> executeQuerySql(String sql) {
  38. 		List<String> result = new ArrayList<String>();
  39. 		try {
  40. 			List<Object[]> requstList = queryRunner.query(getConnection(), sql,
  41. 					new ArrayListHandler(new BasicRowProcessor() {
  42. 						@Override
  43. 						public <Object> List<Object> toBeanList(ResultSet rs,
  44. 								Class<Object> type) throws SQLException {
  45. 							return super.toBeanList(rs, type);
  46. 						}
  47. 					}));
  48. 			for (Object[] objects : requstList) {
  49. 				result.add(objects[0].toString());
  50. 			}
  51. 		} catch (SQLException e) {
  52. 			e.printStackTrace();
  53. 		}
  54. 		return result;
  55. 	}
  56.  
  57. 	@SuppressWarnings("unused")
  58. 	public static void update(String sql, Object... params) {
  59. 		try {
  60. 			Connection connection = getConnection();
  61. 			queryRunner.update(connection, sql, params);
  62. 			connection.close();
  63. 		} catch (SQLException e) {
  64. 			e.printStackTrace();
  65. 		}
  66. 	}
  67.  
  68. 	// 获取连接
  69. 	public static Connection getConnection() throws SQLException {
  70. 		return DriverManager.getConnection(url, user, password);
  71. 	}
  72.  
  73. }

2.storm的topology创建（相当于jobtrack的创建）

  1. import storm.kafka.BrokerHosts;
  2. import storm.kafka.KafkaSpout;
  3. import storm.kafka.SpoutConfig;
  4. import storm.kafka.ZkHosts;
  5. import backtype.storm.Config;
  6. import backtype.storm.LocalCluster;
  7. import backtype.storm.StormSubmitter;
  8. import backtype.storm.generated.AlreadyAliveException;
  9. import backtype.storm.generated.InvalidTopologyException;
  10. import backtype.storm.generated.StormTopology;
  11. import backtype.storm.topology.TopologyBuilder;
  12.  
  13.  
  14. public class CdrTopology {
  15. 	public static void main(String[] args) {
  16. 	TopologyBuilder topologyBuilder = new TopologyBuilder();
  17. 	String KAFKASPOUT = KafkaSpout.class.getSimpleName();
  18. 	String SPLIT_BOLT = SplitBolt.class.getSimpleName();
  19. 	String SAVETOKAFKABOLT = SaveCallLogToKafkaBolt.class.getSimpleName();
  20. 	String SPLIT_BOLT1 = SplitBolt1.class.getSimpleName();
  21. 	String SAVETOKAFKABOLT1 = SaveCallLogToKafkaBolt1.class.getSimpleName();
  22. 	String SAVETOMYSQL = SavaCallLogToMysql.class.getSimpleName();
  23. 	
  24. 	//配置zookeeper
  25. 	BrokerHosts hosts = new ZkHosts("localhost:2181");//指定kafka使用的zk地址
  26. 	String topic = "cdr_log";//主题
  27. 	String zkRoot = "/kafka";//指定一个zk节点，节点不存在会自动创建。[这个节点会创建在storm集群使用的zk中]
  28. 	String id = "123";//groupid
  29. 	SpoutConfig spoutConf = new SpoutConfig(hosts, topic, zkRoot, id);
  30. 	topologyBuilder.setSpout(KAFKASPOUT, new KafkaSpout(spoutConf));
  31. 	//topologyBuilder.setSpout(SPOUT_ID, new KafkaSpout(spoutConf),3);
  32. 	topologyBuilder.setBolt(SPLIT_BOLT1, new SplitBolt1()).shuffleGrouping(KAFKASPOUT);
  33. 	topologyBuilder.setBolt(SAVETOKAFKABOLT1, new SaveCallLogToKafkaBolt1()).shuffleGrouping(SPLIT_BOLT,"calllog");
  34. 	//topologyBuilder.setBolt(SAVETOMYSQL, new SavaCallLogToMysql()).shuffleGrouping(SPLIT_BOLT);
  35. 	
  36. 	StormTopology createTopology = topologyBuilder.createTopology();
  37. 	String simpleName = CdrTopology.class.getSimpleName();
  38. 	Config config = new Config();
  39. 	config.setStatsSampleRate(1D);// 开启精确计数
  40. 	if(args.length==0){
  41. 		LocalCluster localCluster = new LocalCluster();
  42. 		localCluster.submitTopology(simpleName, config, createTopology);
  43. 	}else{
  44. 		try {
  45. 			//config.setNumWorkers(45);
  46. 			config.setMaxSpoutPending(1000);
  47. 			StormSubmitter.submitTopology(simpleName, config, createTopology);
  48. 		} catch (AlreadyAliveException e) {
  49. 			e.printStackTrace();
  50. 		} catch (InvalidTopologyException e) {
  51. 			e.printStackTrace();
  52. 		}
  53. 	}
  54.  
  55. }}

3.作业的分区：

  1. import kafka.producer.Partitioner;
  2. import kafka.utils.VerifiableProperties;
  3.  
  4. /**
  5.  * Created by jason on 2016/11/27.
  6.  */
  7. public class PartitionerDemo implements Partitioner {
  8.  
  9. 	
  10.     private VerifiableProperties verifiableProperties;
  11.  
  12.     public PartitionerDemo(VerifiableProperties verifiableProperties) {
  13.         this.verifiableProperties=verifiableProperties;
  14.     }
  15.  
  16.     public int partition(Object key, int numPartitions) {
  17.  
  18.         String strKey= (String) key;
  19.         //根据userid的hashCode分区
  20.         return strKey.hashCode()%numPartitions;
  21.     }
  22. }

4.数据库各个表的类建立

  1. import java.sql.Connection;
  2. import java.sql.SQLException;
  3. import java.sql.Statement;
  4. import java.util.Map;
  5.  
  6. import cn.com.cintel.storm_siyuan.utils.MyDbUtils;
  7. import clojure.string__init;
  8. import backtype.storm.task.OutputCollector;
  9. import backtype.storm.task.TopologyContext;
  10. import backtype.storm.topology.OutputFieldsDeclarer;
  11. import backtype.storm.topology.base.BaseRichBolt;
  12. import backtype.storm.tuple.Tuple;
  13.  
  14. public class SavaCallLogToMysql extends BaseRichBolt{
  15. 	private OutputCollector collector;
  16. 	private int time=0;
  17.     private String callingnumber="0";
  18.     private String callednumber="0";
  19.     private String callingarea="0";
  20.     private String calledarea="0";
  21. 	int is_land=0;
  22. 	int domain=0;
  23. 	private Connection connection;
  24. 	@Override
  25. 	public void prepare(Map stormConf, TopologyContext context,
  26. 			OutputCollector collector) {
  27. 		// TODO Auto-generated method stub
  28. 		this.collector = collector;
  29. 		this.connection = null;//注意：建议在这使用连接池
  30. 	}
  31.  
  32. 	@Override
  33. 	public void execute(Tuple tuple) {
  34. 		// TODO Auto-generated method stub
  35. 		try {
  36. 			connection = MyDbUtils.getConnection();
  37. 			Statement state = connection.createStatement();
  38. 			time = tuple.getIntegerByField("time");
  39. 			callingnumber = tuple.getStringByField("callingnumber");
  40. 			callednumber = tuple.getStringByField("callednumber");
  41. 			callingarea = tuple.getStringByField("callingarea");
  42. 			calledarea = tuple.getStringByField("calledarea");
  43. 			is_land = tuple.getIntegerByField("is_land");
  44. 			domain = tuple.getIntegerByField("domain");
  45. 			String sql = "insert into calllog(time,callingnumber,callednumber,callingarea,calledarea,is_inland,domain) values("+time+",'"+callingnumber+"',"+"'"+callednumber+"',"+"'"+callingarea+"',"+"'"+calledarea+"',"+is_land+","+domain+")";
  46. 			System.out.println(sql);
  47. 			state.executeUpdate(sql);
  48. 		} catch (SQLException e) {
  49. 			// TODO Auto-generated catch block
  50. 			e.printStackTrace();
  51. 		}finally{
  52. 			if(connection!=null){
  53. 				try {
  54. 					connection.close();
  55. 				} catch (SQLException e) {
  56. 					// TODO Auto-generated catch block
  57. 					e.printStackTrace();
  58. 				}
  59. 			}
  60. 		}
  61. 		
  62. 	}
  63.  
  64. 	@Override
  65. 	public void declareOutputFields(OutputFieldsDeclarer declarer) {
  66. 		// TODO Auto-generated method stub
  67. 		
  68. 	}
  69.  
  70. 	@Override
  71. 	public Map<String, Object> getComponentConfiguration() {
  72. 		// TODO Auto-generated method stub
  73. 		return super.getComponentConfiguration();
  74. 	}
  75.     
  76. }
  1. import java.io.BufferedInputStream;
  2. import java.io.File;
  3. import java.io.FileInputStream;
  4. import java.io.IOException;
  5. import java.io.InputStream;
  6. import java.util.Map;
  7. import java.util.Properties;
  8. import java.util.Random;
  9.  
  10. import kafka.javaapi.producer.Producer;
  11. import kafka.producer.KeyedMessage;
  12. import kafka.producer.ProducerConfig;
  13.  
  14. import org.omg.CORBA.PRIVATE_MEMBER;
  15.  
  16. import backtype.storm.task.OutputCollector;
  17. import backtype.storm.task.TopologyContext;
  18. import backtype.storm.topology.OutputFieldsDeclarer;
  19. import backtype.storm.topology.base.BaseRichBolt;
  20. import backtype.storm.tuple.Tuple;
  21.  
  22. public class SaveCallLogToKafkaBolt extends BaseRichBolt {
  23.     private String topic;
  24. 	private Properties prop;
  25. 	private Producer<String, String> producer;
  26. 	Random random=null;
  27. 	private int time;
  28.     private String callingnumber;
  29.     private String callednumber;
  30.     private String callingarea;
  31.     private String calledarea;
  32. 	int is_land;
  33. 	int domain;
  34. 	@Override
  35. 	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
  36. 		System.out.println("SaveCallLogToKafkaBolt start");
  37. 		topic = "call_log";
  38. 		prop = new Properties();
  39. 		try {
  40. 			//加载producer的配置
  41. 			prop.load(SaveCallLogToKafkaBolt.class.getClassLoader().getResourceAsStream("producer.properties"));
  42. 		} catch (IOException e) {
  43. 			e.printStackTrace();
  44. 		}
  45. 		/*prop.setProperty("metadata.broker.list", "192.168.115.130:9092,192.168.115.132:9092,192.168.115.133:9092");
  46. 		prop.setProperty("partitioner.class", "cn.com.cintel.storm_siyuan.PartitionerDemo");
  47. 		prop.setProperty("producer.type", "sync");
  48. 		prop.setProperty("compression.codec", "none");
  49. 		prop.setProperty("serializer.class", "kafka.serializer.StringEncoder");*/
  50. 		producer = new Producer<>(new ProducerConfig(prop));
  51. 		random=new Random();
  52. 	}
  53. 	@Override
  54. 	public void execute(Tuple tuple) {
  55. 		try
  56.         {		   
  57.             StringBuffer keyedMessage = getKeyedMessage(tuple);
  58.             String msg = keyedMessage.toString();
  59.             //System.out.println(msg);
  60.             Integer tt=random.nextInt(3);
  61.             System.out.println("SaveCallLogToKafkaBolt msg:"+msg);
  62.             producer.send(new KeyedMessage<String, String>(topic,tt.toString(),msg));            
  63.         }
  64.         catch (Exception e)
  65.         {
  66.             // TODO Auto-generated catch block
  67.             e.printStackTrace();
  68.             
  69.         }
  70.  
  71. 	}
  72.  
  73. 	private StringBuffer getKeyedMessage(Tuple tuple) {
  74. 		// TODO Auto-generated method stub
  75. 		time = tuple.getIntegerByField("time");
  76. 		callingnumber = tuple.getStringByField("callingnumber");
  77. 		callednumber = tuple.getStringByField("callednumber");
  78. 		callingarea = tuple.getStringByField("callingarea");
  79. 		calledarea = tuple.getStringByField("calledarea");
  80. 		is_land = tuple.getIntegerByField("is_land");
  81. 		domain = tuple.getIntegerByField("domain");
  82. 		StringBuffer msg = new StringBuffer();
  83. 		msg.append(time);
  84. 		msg.append("|" +callingnumber);
  85. 		msg.append("|" +callednumber);
  86. 		msg.append("|" +callingarea);
  87. 		msg.append("|" +calledarea);
  88. 		msg.append("|" +is_land);
  89. 		msg.append("|" +domain);
  90. 		msg.append("|" +"55");
  91. 		msg.append("|" +"66");
  92.         msg.append("|" +";");
  93. 		return msg; 
  94. 	}
  95. 	@Override
  96. 	public void declareOutputFields(OutputFieldsDeclarer arg0) {
  97. 		// TODO Auto-generated method stub
  98. 		
  99. 	}
  100.  
  101. }

5.storm的bolt处理（bolt：对tuple处理的抽象过程）

  1. import java.util.Map;
  2.  
  3. import backtype.storm.task.OutputCollector;
  4. import backtype.storm.task.TopologyContext;
  5. import backtype.storm.topology.OutputFieldsDeclarer;
  6. import backtype.storm.topology.base.BaseRichBolt;
  7. import backtype.storm.tuple.Fields;
  8. import backtype.storm.tuple.Tuple;
  9. import backtype.storm.tuple.Values;
  10.  
  11. public class SplitBolt1 extends BaseRichBolt{
  12. 	private OutputCollector collector;
  13. 	String log;
  14. 	String[] splited;
  15. 	String time;
  16. 	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
  17. 		// TODO Auto-generated method stub
  18. 		this.collector = arg2;
  19. 		
  20. 	}
  21.  
  22. 	public void execute(Tuple input) {
  23. 		// TODO Auto-generated method stub
  24. 		//time,callingnumber,callednumber,callingarea,calledarea,is_land,domain
  25. 		try {
  26. 			log = new String(input.getBinaryByField("bytes"));
  27. 			splited = log.split("\\|",-1);
  28. 			Message msg = new Message();
  29. 			msg.setTime(Integer.parseInt(splited[0]));
  30. 			msg.setCallingnumber(splited[1]);
  31. 			msg.setCallednumber(splited[2]);
  32. 			msg.setCallingarea(splited[3]);
  33. 			msg.setCalledarea(splited[4]);
  34. 			msg.setIs_land(Integer.parseInt(splited[5]));
  35. 			msg.setDomain(Integer.parseInt(splited[6]));	
  36. 			this.collector.emit("calllog", new Values(msg));
  37. 			this.collector.ack(input);
  38. 		} catch (NumberFormatException e) {
  39. 			// TODO Auto-generated catch block
  40. 			this.collector.fail(input);
  41. 			e.printStackTrace();
  42. 		}
  43. 	}
  44.  
  45. 	public void declareOutputFields(OutputFieldsDeclarer declarer) {
  46. 		// TODO Auto-generated method stub
  47. 		declarer.declareStream("calllog",new Fields("calllog"));
  48. 	}
  49.  
  50. }

对于spark和hive的完善，请看下篇。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/362708?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号