首页 文章 精选 留言 我的

精选列表

搜索[数据源],共7895篇文章
优秀的个人博客,低调大师

Spark连接JDBC数据源

在实际的项目环境中,成熟的技术体系对关系型数据库的依赖远远超过hdfs,而且对大数据运算的结果,实践中也倾向于保存到数据库中,以便进行多种方式的可视化。所以本次实践主要完成spark从mysql中读取和写入数据。一般这个操作有两种方式,一种是自己建立jdbc连接,像一般数据库操作一样的写法,一种就是利用spark自带的jdbc操作函数。 首先要把mysql jdbc connector的jar包上传到集群中每台机器的spark/jars目录,这是一个讨巧的办法,因为spark运行之前一定把这里面所有的jar都加到CALSS_PATH里面去了。 通过spark.read.jdbc读取出来的返回值是DataFrame,如下代码所示。` val rfidCardMap = spark.read.jdbc(mysqlHelper.DB_URL_R,"t_rfid_card",Array("org_id="+ ORG_ID), mysqlHelper.PROPERTIES).map(row => { (row.getAs[String]("card_id"), row.getAs[String]("card_label")) }).rdd.collect() toMap` 此函数需要传入参数依次为:数据库连接url,表名,过滤条件表达式列表,带有用户名密码信息的属性对象。读取了数据之后,形成一个(String,String)对象返回。这里有两个要注意的: getAs的类型必须和数据库中列的类型严格匹配 返回元组类型的对象比返回自定义类的对象写法要轻松一些。如果是返回自定义类的对象,编译会出错,一般说法是语句之前加入import spark.implicits._会有效,但未必见得。尚待进一步探索。 如下是一个比较复杂的解析处理代码示例。` val teamWeightMapRDD = dfMedicalWaste.map(row => { (rfidCardMap.get(row.getAs[String]("team_id")) toString, sdf.format(new Date(row.getAs[Timestamp]("rec_ts").getTime)) toInt, row.getAs[Double]("mw_weight")) }).rdd.cache()` 这里sdf就是java里面常用的SimpleDateFormat,它把一个时间戳字段转化成了6个长度的整型。 处理完成后,将结果回写数据库时采用的是本地jdbc连接写法,这块内容很普通了。 这次实践有个特别清晰的理解就是scala的类型推断,由于要统计某个地点一段时间之内的产量总和、平均产量、最大和最小单位时间产量,使用到了DoubleRDDFunctions,代码如下:` val weightArrayRDD = teamWeightMapRDD.filter(teamWeight => { teamWeight._1 == teamName && teamWeight._2 >= week._1 && teamWeight._2 < week._2 }).map(teamWeight => { (teamWeight._2, teamWeight._3) }).reduceByKey((a, b) => a + b ).map(item => { item._2 }).cache()` 使用的时候如下:` line.append(weightArrayRDD sum).append("\t") line.append(weightArrayRDD mean).append("\t") line.append(weightArrayRDD max).append("\t") line.append(weightArrayRDD min).append("\t")` scala会根据返回值类型进行类型推断,从而匹配可以使用的函数,同样是RDD或者DataFram,包含的类型不同,可以使用的函数也不同,这一切都是透明的。

优秀的个人博客,低调大师

SpringBoot整合MongoDB多数据源

依赖 // build.gradle implementation 'org.springframework.boot:spring-boot-starter-data-mongodb' 配置文件 # application.yml spring: data: mongodb: primary: uri: mongodb://localhost:27017/db1 secondary: uri: mongodb://localhost:27017/db2 主数据库配置 // PrimaryMongoConfig.java package com.fengwenyi.springboot_mongo_multi_source.config; import com.mongodb.MongoClientURI; import org.springframework.boot.autoconfigure.mongo.MongoProperties; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Primary; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.data.mongodb.repository.config.EnableMongoRepositories; /** * MongoDB Primary Config * @author Erwin Feng * @since 2019-07-01 17:12 */ @Configuration @EnableMongoRepositories(basePackages = "com.fengwenyi.springboot_mongo_multi_source.primary", mongoTemplateRef = "primaryMongoTemplate") public class PrimaryMongoConfig { @Bean @Primary @ConfigurationProperties(prefix="spring.data.mongodb.primary") public MongoProperties primaryMongoProperties() { return new MongoProperties(); } @Primary @Bean(name = "primaryMongoTemplate") public MongoTemplate primaryMongoTemplate() throws Exception { return new MongoTemplate(primaryFactory(primaryMongoProperties())); } @Bean @Primary public MongoDbFactory primaryFactory(MongoProperties mongoProperties) throws Exception { return new SimpleMongoDbFactory(new MongoClientURI(primaryMongoProperties().getUri())); } } 副数据库配置 // SecondaryMongoConfig.java package com.fengwenyi.springboot_mongo_multi_source.config; import com.mongodb.MongoClientURI; import org.springframework.boot.autoconfigure.mongo.MongoProperties; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Primary; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.data.mongodb.repository.config.EnableMongoRepositories; /** * MongoDB Secondary Config * @author Erwin Feng * @since 2019-07-01 17:12 */ @Configuration @EnableMongoRepositories(basePackages = "com.fengwenyi.springboot_mongo_multi_source.secondary", mongoTemplateRef = "secondaryMongoTemplate") public class SecondaryMongoConfig { @Bean @ConfigurationProperties(prefix="spring.data.mongodb.secondary") public MongoProperties secondaryMongoProperties() { return new MongoProperties(); } @Bean(name = "secondaryMongoTemplate") public MongoTemplate secondaryMongoTemplate() throws Exception { return new MongoTemplate(secondaryFactory(secondaryMongoProperties())); } @Bean public MongoDbFactory secondaryFactory(MongoProperties mongoProperties) throws Exception { return new SimpleMongoDbFactory(new MongoClientURI(secondaryMongoProperties().getUri())); } } 用户实体,副 // User.java package com.fengwenyi.springboot_mongo_multi_source.secondary.entity; import lombok.Data; import lombok.experimental.Accessors; import org.springframework.data.annotation.Id; import org.springframework.data.mongodb.core.mapping.Document; import java.io.Serializable; import java.time.Instant; /** * 用户 * @author Erwin Feng * @since 2019-07-01 17:15 */ @Data @Accessors(chain = true) @Document(collection = "t_user") public class User implements Serializable { private static final long serialVersionUID = -7229906944062898852L; /** ID */ @Id private String id; /** 用户名 */ private String username; /** 年龄 */ private Integer age; /** 注册时间 */ private Instant registerTime; } 用户查询仓库,副 // UserRepository.java package com.fengwenyi.springboot_mongo_multi_source.secondary.repository; import com.fengwenyi.springboot_mongo_multi_source.secondary.entity.User; import org.springframework.data.mongodb.repository.MongoRepository; import java.util.List; /** * 用户 * @author Erwin Feng * @since 2019-07-01 17:18 */ public interface UserRepository extends MongoRepository<User, String> { /** * 通过用户名查询 * @param username 用户名 * @return */ List<User> findAllByUsername(String username); } 登录日志实体,主 // LoginLog.java package com.fengwenyi.springboot_mongo_multi_source.primary.entity; import lombok.Data; import lombok.experimental.Accessors; import org.springframework.data.annotation.Id; import java.io.Serializable; import java.time.Instant; /** * 登录日志 * @author Erwin Feng * @since 2019-07-01 17:18 */ @Data @Accessors(chain = true) public class LoginLog implements Serializable { private static final long serialVersionUID = -6694661682102504919L; /** ID */ @Id private String id; /** 用户ID */ private String uid; /** 用户名 */ private String username; /** 登录时间 */ private Instant loginTime; } 登录日志查询仓库,主 // LoginLogRepository.java package com.fengwenyi.springboot_mongo_multi_source.primary.repository; import com.fengwenyi.springboot_mongo_multi_source.primary.entity.LoginLog; import org.springframework.data.mongodb.repository.MongoRepository; /** * 登录日志 * @author Erwin Feng * @since 2019-07-01 17:21 */ public interface LoginLogRepository extends MongoRepository<LoginLog, String> { } 初始化 // InitController.java package com.fengwenyi.springboot_mongo_multi_source.controller; import com.fengwenyi.springboot_mongo_multi_source.secondary.entity.User; import com.fengwenyi.springboot_mongo_multi_source.secondary.repository.UserRepository; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RestController; import javax.annotation.PostConstruct; import java.time.Instant; import java.util.ArrayList; import java.util.List; /** * 初始化工具类 * @author Erwin Feng * @since 2019-07-01 17:24 */ @RestController public class InitController { /** [mongo] 用户 */ @Autowired private UserRepository userRepository; @PostConstruct public void init() { List<User> all = userRepository.findAll(); if (all.size() > 0) return; userRepository.save(new User().setUsername("Zhangsan").setAge(20).setRegisterTime(Instant.now())); List<User> users = new ArrayList<>(); User u1 = new User().setUsername("u1").setAge(19).setRegisterTime(Instant.now()); User u2 = new User().setUsername("u2").setAge(20).setRegisterTime(Instant.now()); User u3 = new User().setUsername("u3").setAge(10).setRegisterTime(Instant.now()); users.add(u1); users.add(u2); users.add(u3); userRepository.saveAll(users); } } 测试代码 // TestController.java package com.fengwenyi.springboot_mongo_multi_source.controller; import com.fengwenyi.springboot_mongo_multi_source.primary.entity.LoginLog; import com.fengwenyi.springboot_mongo_multi_source.primary.repository.LoginLogRepository; import com.fengwenyi.springboot_mongo_multi_source.secondary.entity.User; import com.fengwenyi.springboot_mongo_multi_source.secondary.repository.UserRepository; import net.iutil.ApiResult; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Query; import org.springframework.util.StringUtils; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.time.Instant; import java.util.List; /** * 测试 * @author Erwin Feng * @since 2019-07-01 17:28 */ @RestController @RequestMapping("/test") public class TestController { /** [mongo] 用户 */ @Autowired private UserRepository userRepository; /** [mongo] 登录日志 */ @Autowired private LoginLogRepository loginLogRepository; /** [mongo] */ @Autowired private MongoTemplate mongoTemplate; /** * 登录 * @param username * @return */ @GetMapping("/login") public ApiResult login(String username) { if (StringUtils.isEmpty(username)) return ApiResult.error().setMsg("用户名不能为空"); List<User> users = userRepository.findAllByUsername(username); if (users.size() == 1) { // 记录日志 loginLogRepository.save(new LoginLog().setUid(users.get(0).getId()).setUsername(username).setLoginTime(Instant.now())); return ApiResult.success(); } if (users.size() == 0) return ApiResult.error().setMsg("用户名查询失败"); return ApiResult.error().setMsg("用户异常"); } /** * 登录日志 * @return */ @GetMapping("/login-log") public ApiResult loginLog() { Query query = new Query(); List<LoginLog> loginLogs = mongoTemplate.find(query, LoginLog.class); return ApiResult.success(loginLogs); } } 测试用户登录 GET http://localhost:8080/test/login?username=Zhangsan 响应: { "code": 0, "msg": "Success" } 测试登录日志 GET http://localhost:8080/test/login-log 响应: { "code": 0, "msg": "Success", "data": [ { "id": "5d19d7f5cede54c46b6b20c5", "uid": "5d19d560cede54c45701e12a", "username": "Zhangsan", "loginTime": "2019-07-01T09:52:53.447Z" }, { "id": "5d19da82cede54c46f77579a", "uid": "5d19d560cede54c45701e12a", "username": "Zhangsan", "loginTime": "2019-07-01T10:03:46.496Z" }, { "id": "5d19df5fcede54c46f77579b", "uid": "5d19d560cede54c45701e12a", "username": "Zhangsan", "loginTime": "2019-07-01T10:24:31.272Z" }, { "id": "5d19df6acede54c46f77579c", "uid": "5d19d560cede54c45701e12b", "username": "u1", "loginTime": "2019-07-01T10:24:42.199Z" }, { "id": "5d19df6dcede54c46f77579d", "uid": "5d19d560cede54c45701e12d", "username": "u3", "loginTime": "2019-07-01T10:24:45.421Z" } ] } 代码 springboot-mongo-multi-source

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

用户登录
用户注册