基于Spark2.x新闻网大数据实时分析可视化系统项目-北京尧图网络

基于Spark2.x新闻网大数据实时分析可视化系统项目

本次项目是基于企业大数据经典案例项目大数据日志分析全方位、全流程讲解大数据项目的业务分析、技术选型、架构设计、集群规划、安装部署、整合继承与开发和web可视化交互设计。项目代码托管于github大家可以自行下载。一、业务需求分析捕获用户浏览日志信息实时分析前20名流量最高的新闻话题实时统计当前线上已曝光的新闻话题统计哪个时段用户浏览量最高二、系统架构图设计三、系统数据流程设计四、集群资源规划设计五、步骤详解考虑到实际情况本人集群配置共三个节点node5、node6、node7。1. Zookeeper分布式集群部署参考博客2. Hadoop2.X HA架构与部署参考博客3. HBase分布式集群部署与设计参考博客4. Kafka分布式集群部署参考博客5. Flume部署及数据采集准备参考博客node6与node7中flume数据采集到node5中而且node6和node7的flume配置文件大致相同node7中将a2改为a3如下a2.sources r1 a2.sinks k1 a2.channels c1 a2.sources.r1.type exec a2.sources.r1.command tail -F /opt/data/weblog-flume.log a2.sources.r1.channels c1 a2.channels.c1.type memory a2.channels.c1.capacity 1000 a2.channels.c1.transactionCapacity 1000 a2.channels.c1.keep-alive 5 a2.sinks.k1.type avro a2.sinks.k1.channel c1 a2.sinks.k1.hostname node5 a2.sinks.k1.port 55556. FlumeHBaseKafka集成与开发1. 下载Flume源码并导入Idea开发工具1将apache-flume-1.7.0-src.tar.gz源码下载到本地解压2通过idea导入flume源码打开idea开发工具选择File——》Open然后找到flume源码解压文件选中flume-ng-hbase-sink点击ok加载相应模块的源码。2. 官方flume与hbase集成的参数介绍3. 下载日志数据并分析到搜狗实验室下载用户查询日志1介绍搜索引擎查询日志库设计为包括约1个月(2008年6月)Sogou搜索引擎部分网页查询需求及用户点击情况的网页查询日志数据集合。为进行中文搜索引擎用户行为分析的研究者提供基准研究语料。2格式说明数据格式为:访问时间\t用户ID\t[查询词]\t该URL在返回结果中的排名\t用户点击的顺序号\t用户点击的URL其中用户ID是根据用户使用浏览器访问搜索引擎时的Cookie信息自动赋值即同一次使用浏览器输入的不同查询对应同一个用户ID4. node5聚合节点与HBase和Kafka的集成配置node5通过flume接收node6与node7中flume传来的数据并将其分别发送至hbase与kafka中配置内容如下a1.sources r1 a1.channels kafkaC hbaseC a1.sinks kafkaSink hbaseSink a1.sources.r1.type avro a1.sources.r1.channels hbaseC kafkaC a1.sources.r1.bind node5 a1.sources.r1.port 5555 a1.sources.r1.threads 5 #****************************flume hbase****************************** a1.channels.hbaseC.type memory a1.channels.hbaseC.capacity 10000 a1.channels.hbaseC.transactionCapacity 10000 a1.channels.hbaseC.keep-alive 20 a1.sinks.hbaseSink.type asynchbase a1.sinks.hbaseSink.table weblogs a1.sinks.hbaseSink.columnFamily info a1.sinks.hbaseSink.serializer org.apache.flume.sink.hbase.KfkAsyncHbaseEventSerializer a1.sinks.hbaseSink.channel hbaseC a1.sinks.hbaseSink.serializer.payloadColumn datetime,userid,searchname,retorder,cliorder,cliurl #****************************flume kafka****************************** a1.channels.kafkaC.type memory a1.channels.kafkaC.capacity 10000 a1.channels.kafkaC.transactionCapacity 10000 a1.channels.kafkaC.keep-alive 20 a1.sinks.kafkaSink.channel kafkaC a1.sinks.kafkaSink.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.kafkaSink.brokerList node5:9092,node6:9092,node7:9092 a1.sinks.kafkaSink.topic weblogs a1.sinks.kafkaSink.zookeeperConnect node5:2181,node6:2181,node7:2181 a1.sinks.kafkaSink.requiredAcks 1 a1.sinks.kafkaSink.batchSize 1 a1.sinks.kafkaSink.serializer.class kafka.serializer.StringEncoder5. 对日志数据进行格式处理1将文件中的tab更换成逗号cat weblog.log|tr \t , weblog.log2将文件中的空格更换成逗号cat weblog2.log|tr , weblog.log6. 自定义SinkHBase程序设计与开发1模仿SimpleAsyncHbaseEventSerializer自定义KfkAsyncHbaseEventSerializer实现类修改一下代码即可。Override public ListPutRequest getActions() { ListPutRequest actions new ArrayListPutRequest(); if (payloadColumn ! null) { byte[] rowKey; try { /*---------------------------代码修改开始---------------------------------*/ // 解析列字段 String[] columns new String(this.payloadColumn).split(,); // 解析flume采集过来的每行的值 String[] values new String(this.payload).split(,); for(int i0;i columns.length;i){ byte[] colColumn columns[i].getBytes(); byte[] colValue values[i].getBytes(Charsets.UTF_8); // 数据校验字段和值是否对应 if(colColumn.length ! colValue.length) break; // 时间 String datetime values[0].toString(); // 用户id String userid values[1].toString(); // 根据业务自定义Rowkey rowKey SimpleRowKeyGenerator.getKfkRowKey(userid,datetime); // 插入数据 PutRequest putRequest new PutRequest(table, rowKey, cf, colColumn, colValue); actions.add(putRequest); /*---------------------------代码修改结束---------------------------------*/ } } catch (Exception e) { throw new FlumeException(Could not get row key!, e); } } return actions; }2在SimpleRowKeyGenerator类中根据具体业务自定义Rowkey生成方法public static byte[] getKfkRowKey(String userid, String datetime) throws UnsupportedEncodingException { return (userid - datetime - String.valueOf(System.currentTimeMillis())).getBytes(UTF8); }7. 自定义编译程序打jar包1在idea工具中选择File——》ProjectStructrue2左侧选中Artifacts然后点击右侧的号最后选择JAR——》From modules with dependencies3然后直接点击ok4删除其他依赖包只把flume-ng-hbase-sink打成jar包就可以了。5然后依次点击applyok6点击build进行编译会自动打成jar包7到项目的apache-flume-1.7.0-src\flume-ng-sinks\flume-ng-hbase-sink\classes\artifacts\flume_ng_hbase_sink_jar目录下找到刚刚打的jar包8将打包名字替换为flume自带的包名flume-ng-hbase-sink-1.7.0.jar 然后上传至flume/lib目录下覆盖原有的jar包即可。7. 数据采集/存储/分发完整流程测试1. 在idea开发工具中构建weblogs项目编写数据生成模拟程序。package main.java; import java.io.*; public class ReadWrite { static String readFileName; static String writeFileName; public static void main(String args[]){ readFileName args[0]; writeFileName args[1]; try { // readInput(); readFileByLines(readFileName); }catch(Exception e){ } } public static void readFileByLines(String fileName) { FileInputStream fis null; InputStreamReader isr null; BufferedReader br null; String tempString null; try { System.out.println(以行为单位读取文件内容一次读一整行); fis new FileInputStream(fileName);// FileInputStream // 从文件系统中的某个文件中获取字节 isr new InputStreamReader(fis,GBK); br new BufferedReader(isr); int count0; while ((tempString br.readLine()) ! null) { count; // 显示行号 Thread.sleep(300); String str new String(tempString.getBytes(UTF8),GBK); System.out.println(row:counttempString); method1(writeFileName,tempString); //appendMethodA(writeFileName,tempString); } isr.close(); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } finally { if (isr ! null) { try { isr.close(); } catch (IOException e1) { } } } } public static void method1(String file, String conent) { BufferedWriter out null; try { out new BufferedWriter(new OutputStreamWriter( new FileOutputStream(file, true))); out.write(\n); out.write(conent); } catch (Exception e) { e.printStackTrace(); } finally { try { out.close(); } catch (IOException e) { e.printStackTrace(); } } } }2. 参照前面idea工具项目打包方式将该项目打成weblogs.jar包然后上传至bigdata-pro01.kfk.com节点的/opt/jars目录下目录需要提前创建3. 将weblogs.jar分发到另外两个节点node6和node71在另外两个节点上分别创建/opt/jars目录mkdir /opt/jars2将weblogs.jar分发到另外两个节点scp weblogs.jar node6:/opt/jars/ scp weblogs.jar node7:/opt/jars/4. 编写运行模拟程序的shell脚本在node6和node7节点的/opt/datas目录下创建weblog-shell.sh脚本。内容为#/bin/bash echo start log...... #第一个参数是原日志文件第二个参数是日志生成输出文件 java -jar /opt/jars/weblogs.jar /opt/datas/weblog.log /opt/datas/weblog-flume.log修改weblog-shell.sh可执行权限chmod 777 weblog-shell.sh5. 编写启动flume服务程序的shell脚本在各节点的flume安装目录下编写flume启动脚本flume-kfk-start.sh。下面是node5中的配置写法node6与node7中将a1分别改为a2和a3。#/bin/bash echo flume-1 start ...... bin/flume-ng agent --conf conf -f conf/flume-conf.properties -n a1 -Dflume.root.loggerINFO,console6. 编写Kafka Consumer执行脚本kfk-test-consumer.sh。#/bin/bash echo kfk-kafka-consumer.sh start...... bin/kafka-console-consumer.sh --zookeeper node5:2181,node6:2181,node7:2181 --from-beginning --topic weblogs7. 将kfk-test-consumer.sh脚本分发另外两个节点scp kfk-test-consumer.sh node6:/opt/modules/kakfa_2.11-0.8.2.1/ scp kfk-test-consumer.sh node7:/opt/modules/kakfa_2.11-0.8.2.1/8. 启动模拟程序并测试在node6节点启动日志产生脚本模拟产生日志是否正常。/opt/datas/weblog-shell.sh9. 启动数据采集所有服务1启动Zookeeper服务2启动hdfs服务3启动HBase服务创建hbase业务表create weblogs,info4启动Kafka服务并创建业务数据topicbin/kafka-server-start.sh config/server.properties bin/kafka-topics.sh --create --zookeeper node5:2181,node6:2181,node7:2181 --topic weblogs --partitions 1 --replication-factor 310. 完成数据采集全流程测试1在node5节点上启动flume聚合脚本将采集的数据分发到Kafka集群和hbase集群。./flume-kfk-start.sh2在node6和node7节点上完成数据采集。1使用shell脚本模拟日志产生cd /opt/datas/ ./weblog-shell.sh2启动flume采集日志数据发送给聚合节点./flume-kfk-start.sh3启动Kafka Consumer查看flume日志采集情况bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic weblogs --from-beginning4查看hbase数据写入情况./hbase-shell count weblogs8. MySQL安装参考博客9. Hive与HBase集成进行数据分析1. Hive安装本地mysql模式参考博客2. Hive与HBase集成1在hive-site.xml文件中配置Zookeeperhive通过这个参数去连接HBase集群。property namehbase.zookeeper.quorum/name valuenode5,node6,node7/value /property2将hbase的9个包拷贝到hive/lib目录下。如果是CDH版本已经集成好不需要导包。export HBASE_HOME/opt/modules/hbase-0.98.6-cdh5.3.0 export HIVE_HOME/opt/modules/hive-0.13.1/lib ln -s $HBASE_HOME/lib/hbase-server-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-server-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/hbase-client-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-client-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/hbase-protocol-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-protocol-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/hbase-it-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-it-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/htrace-core-2.04.jar$HIVE_HOME/lib/htrace-core-2.04.jar ln -s $HBASE_HOME/lib/hbase-hadoop2-compact-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-hadoop2-compact-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/hbase-hadoop-compact-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-hadoop-compact-0.98.6-cdh5.3.0.jar ln -s $HBASE_HOME/lib/high-scale-lib-1.1.1.jar $HIVE_HOME/lib/high-scale-lib-1.1.1.jar ln -s $HBASE_HOME/lib/hbase-common-0.98.6-cdh5.3.0.jar $HIVE_HOME/lib/hbase-common-0.98.6-cdh5.3.0.jar3在hive中创建与hbase集成的外部表CREATE EXTERNAL TABLE weblogs( id string, datetime string, userid string, searchname string, retorder string, cliorder string, cliurl string ) STORED BY org.apache.hadoop.hive.hbase.HBaseStorageHandler WITH SERDEPROPERTIES(hbase.columns.mapping :key,info:datetime,info:userid,info:searchname,info:retorder,info:cliorder,info:cliurl) TBLPROPERTIES(hbase.table.nameweblogs); #查看hbase数据记录 select count(*) from weblogs;4hive中beeline和hiveserver2的使用# 启动hiveserver2 bin/hiveserver2 # 启动beeline bin/beeline # 连接hive2服务 !connect jdbc:hive2//node7.kfk.com:10000 # 查看表 show tables; # 查看前10条数据 select * from weblogs limit 10;10. Cloudera HUE大数据可视化分析参考博客11. Spark2.X环境准备、编译部署及运行参考博客搭建spark集群参考博客12. Spark SQL快速离线数据分析1. Spark SQL 与Hive集成spark-shell参考博客2. Spark SQL 与Hive集成spark-sql参考博客3. Spark SQL之ThriftServer和beeline使用参考博客4. Spark SQL与MySQL集成参考博客5. Spark SQL与HBase集成参考博客13. Structured Streaming业务数据实时分析1. Structured Streaming与kafka集成1Structured Streaming是Spark2.2.0新推出的要求kafka的版本0.10.0及以上。集成时需将如下的包拷贝到Spark的jar包目录下。kafka_2.11-0.10.1.0.jar kafka-clients-0.10.1.0.jar spark-sql-kafka-0-10_2.11-2.2.0.jar spark-streaming-kafka-0-10_2.11-2.1.0.jar2与kafka集成代码val df spark .readStream .format(kafka) .option(kafka.bootstrap.servers, node5:9092) .option(subscribe, weblogs) .load() import spark.implicits._ val lines df.selectExpr(CAST(value AS STRING)).as[String]2. Structured Streaming与MySQL集成1mysql创建相应的数据库和数据表用于接收数据create database test; use test; CREATE TABLE webCount ( titleName varchar(255) CHARACTER SET utf8 DEFAULT NULL, count int(11) DEFAULT NULL ) ENGINEInnoDB DEFAULT CHARSETutf8;2与mysql集成代码val url jdbc:mysql://node5:3306/test val usernameroot val password1234 val writer new JDBCSink(url,username,password) val query titleCount.writeStream .foreach(writer) .outputMode(update) .trigger(ProcessingTime(5 seconds)) .start()3. Structured Streaming向mysql数据库写入中文乱码解决修改数据库文件my.cnflinux下[client] socket/var/lib/mysql/mysql.sock //添加 default-character-setutf8 //添加 [mysqld] character-set-serverutf8 //添加 datadir/var/lib/mysql socket/var/lib/mysql/mysql.sock usermysql # Disabling symbolic-links is recommended to prevent assorted security risks symbolic-links0 [mysqld_safe] log-error/var/log/mysqld.log pid-file/var/run/mysqld/mysqld.pid14. 大数据Web可视化分析系统开发1. 基于业务需求的WEB系统设计具体参照代码2. 基于Echart框架的页面展示层开发1echart、JQuery下载2页面效果图选取及代码实现3. 工程编译并打包发布参照之前将的idea打包方式将spark web项目打包发布。4. 启动各个服务1启动zookeeper zkServer.sh start2启动hadoop start-all.sh3启动hbase start-hbase4启动mysql service mysqld start5node6node7启动flume flume-kfk-start.sh将数据发送到node5中6node5启动flume flume-kfk-start.sh将数据分别传到hbase和kafka中7启动kafka-0.10(最好三台都启动不然易出错)bin/kafka-server-start.sh config/server.properties kafka.log 21 8启动node6(node7)中的脚本weblog-shell.sh9启动 StructuredStreamingKafka来从kafka中取得数据处理后存到mysql中10启动web项目sparkStu该项目会从mysql数据库中读取数据展示到页面5. 最终项目运行效果

RELATED

相关推荐

零碳校园能耗管理平台建设：从电表到碳报告的全链路实践

零碳校园能耗管理平台建设：从电表到碳报告的全链路实践

简介：这是一份面向智慧校园建设者与碳中和规划人员的零碳绿色校园能耗管理平台建设方案PPT。内容基于当前智慧校园基础层建设背景，从两部委绿色学校创建政策与3060双碳目标切入，梳理零碳校园“三步走”建设路径，并围绕碳资产及能效…

📅 2026/10/22 6:32:10

XSS深度审计实战：从上下文分析到CSP绕过策略

XSS深度审计实战：从上下文分析到CSP绕过策略

很多做安全的朋友习惯在漏洞点放一个<script>alert(1)</script>，弹窗成功就“收获”，没弹窗就觉得这里没有 XSS。这个习惯做功能验证还行，做深度审计容易漏掉真正的风险。Alert(1)只是探测器，不是武器库——真正要回答…

📅 2026/10/22 6:32:10

创新创业大赛路演PPT改造指南：叙事框架、内容替换与答辩避坑

创新创业大赛路演PPT改造指南：叙事框架、内容替换与答辩避坑

简介：这份创新创业大赛创意PPT以“企业效率管理平台”为核心，适合创业团队、中小企业管理者及高校教学参考，清晰阐述了如何借助一体化平台提升内部管理与外部销售效率。压缩包内仅有1个PPT格式文件，体积约438KB，短小精…

📅 2026/10/22 6:32:10

MORE NEWS

更多资讯

OpenCore Legacy Patcher 3.0.0 上手指南：老 Intel Mac 安装最新 macOS 全流程

OpenCore Legacy Patcher 3.0.0 上手指南：老 Intel Mac 安装最新 macOS 全流程【免费下载链接】OpenCore-Legacy-Patcher Experience macOS just like before 项目地址: https://gitcode.com/GitHub_Trending/op/OpenCore-Legacy-Patcher OpenCore Legacy P…

RevokeMsgPatcher 微信防撤回补丁：改几个字节，撤回消息本地留底

RevokeMsgPatcher 微信防撤回补丁：改几个字节，撤回消息本地留底【免费下载链接】RevokeMsgPatcher :trollface: A hex editor for WeChat/QQ/TIM - PC版微信/QQ/TIM防撤回补丁（我已经看到了，撤回也没用了） 项目地址…

10 分钟上手 AxisRL：从 Docker 环境到第一条 GRPO 训练命令的完整新手指南

【免费下载链接】axrl AxisRL is an agentic RL post-training framework built on SGLang rollout, Megatron training, and real-world agent workflows. 项目地址： https://gitcode.com/gh_mirrors/ax/axrl 点击查看免费下载 AxisRL 是一个面向 agentic RL 的 …

焊盘掉了别直接扔板：这招飞线修复，省了几千换板成本

干修板这行，谁没碰过掉焊盘的糟心事？拆芯片没等锡化透就敢硬撬，老板子常年高温烤得铜皮发脆，稍微用点力，“啪”一下焊盘连带表层走线掀起来一小块。新手一瞅直接慌神，完了，板废了，要…

免费换回Win10任务栏：ExplorerPatcher教程

免费换回Win10任务栏：ExplorerPatcher教程【免费下载链接】ExplorerPatcher This project aims to enhance the working environment on Windows 项目地址: https://gitcode.com/GitHub_Trending/ex/ExplorerPatcher ExplorerPatcher是免费开源（…

浏览器相关工具推荐：别再只会装一个 Chrome，真正好用的是这些插件和增强玩法

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

TODAY

今日更新

电商智能运营体系从0到1：架构设计、落地步骤与避坑指南

前两年我们团队接了一个电商平台的运营中台项目，目标很直白：把过去靠运营拍脑袋做的选品、定价、发券、推活动这些事，逐步用数据算法和自动化流程代替。项目名字内部就叫“电商智能运营体系”。最近有不少朋友问我当时是怎么设计的、踩过哪些…

压缩空气储能与热网热惯性联合优化调度：热电联产系统建模与求解

前几年做模拟项目X时，我遇到一个很典型的调度困境：系统里有一台热电联产机组和一片集中供热管网，冬季夜间电负荷低、热负荷却很高，机组被“以热定电”的约束绑得死死的，电力调峰完全腾不出手，只能眼睁睁看着…

若依框架AI代码生成不稳定？用Agent Skill三层结构+五步法搞定

最近一段时间，我一直想把后台管理系统里那些机械重复的增删改查代码交给AI去写。框架用的是若依，按理说这类业务代码模式非常固定，应该是AI最擅长的事情。可真试下来却不是那么回事——你让AI生成一个用户管理模块，它给出的Contro…

THIS WEEK

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

简介：这份PDF报告聚焦2025年金融服务业中国AI现状与趋势，面向金融科技从业者、AI战略规划人员及关注行业数字化转型的研究者，帮助读者把握生成式AI在银行、支付、交易等场景中的落地路径与投资方向。资源包内含1个PDF文件，大小约1…

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

基于GMM的语音识别系统原理与MATLAB实现详解

说到基于高斯混合模型（GMM）的语音识别系统，很多人第一反应就是一堆数学公式，觉得门槛很高。但如果你真的在MATLAB里把它落地跑通，会发现核心其实就是三件事：把声音变成特征向量，用GMM给每个类别…

THIS MONTH

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

读完文章，想聊聊您的网站？

告诉我们您的行业与需求，资深顾问一对一梳理方案与报价，全程免费。

免费获取方案 400-888-6620

📞 💬