spark-sql(jdbc)本地模式导出csv或Excel文件

这篇具有很好参考价值的文章主要介绍了spark-sql(jdbc)本地模式导出csv或Excel文件。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.spark-sql读取数据jdbc或者hive数据本地模式导出依赖

        <dependency>
            <groupId>org.apache.poi</groupId>
            <artifactId>poi-ooxml</artifactId>
            <version>5.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.poi</groupId>
            <artifactId>poi</artifactId>
            <version>5.2.1</version>
        </dependency>
        <dependency>
            <groupId>jxl</groupId>
            <artifactId>jxl</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.11.0</version>
        </dependency>

注意:当前excel和commons-io版本都是较较新版本,而commons-io在spark的jars安装目录下也在commons-io的包,如版本冲突,找不到org\apache\commons\io\output\ByteArrayOutputStream.class。如果spark的是2.4或者更低版本,则找不到org\apache\commons\io\output\UnsynchronizedByteArrayOutputStream.class,请同步spark的版本的当前应用包的即可!

2.spark-sql导出excel或者csv

       当前是使用spark的本地聚集方法,Iterator<Row> localIterator = dataset.toLocalIterator();把数据聚集到本地再导出,纯手工实现,可灵活调,不适合大数据集批量导出,在结果集不大的情况可使用此方法导出,仅供参考!文章来源地址https://www.toymoban.com/news/detail-430711.html

2.1导出代码

    /**
     * 生成文件导出
     *
     * @param sparkSession sparkSession
     * @param paramsMap 参数
     */
    public static void sqlAccessFileExport(SparkSession sparkSession, Map<String, Object> paramsMap) {

        Long downloadTaskId = MapParamUtil.getLongValue(paramsMap, "downloadTaskId");
        String taskName = MapParamUtil.getStringValue(paramsMap, "taskName");
        String fileType = MapParamUtil.getStringValue(paramsMap, "fileType");
        String waterMark = MapParamUtil.getStringValue(paramsMap, "waterMark");
        String tmpDir = MapParamUtil.getStringValue(paramsMap, "tmpDir");
        String receiveClient = MapParamUtil.getStringValue(paramsMap, "receiveClient");
        Map<String, Object> metaData = (Map<String, Object>) paramsMap.get("metaData");

        // 文件名称
        String fileName = UUID.randomUUID().toString().concat(".").concat(fileType);
        if (!tmpDir.endsWith("/")) {
            tmpDir = tmpDir.concat("/");
        }
        String fileLocation = tmpDir.concat(fileName);

        Map<String, Object> returnMap = new HashMap<String, Object>(10);
        returnMap.put("downloadTaskId", downloadTaskId);
        returnMap.put("method", "sqlAccessFileExportResult");

        try {

            long start = System.currentTimeMillis();

            Dataset<Row> dataset = SqlAccessFile.getDataset(sparkSession, paramsMap);

            // 如果是excel文件,创建方式不同
            if ("xlsx".equalsIgnoreCase(fileType) || "xls".equalsIgnoreCase(fileType)) {
                SqlAccessFile.createExcel(dataset, fileLocation, taskName, waterMark, metaData);
            }
            else {
                // 创建csv文件
                SqlAccessFile.createCsvOrTxt(dataset, fileLocation, metaData);
            }

            long end = System.currentTimeMillis();

            returnMap.put("resultCode", KeyValues.SUCCESS);
            returnMap.put("resultDesc", "PTO label run success:" + (end - start) + "ms");
            returnMap.put("total", dataset.count());
            returnMap.put("fileLocation", fileLocation);
            SendMessageUtil.sendMessage(receiveClient, returnMap);
        }
        catch (Exception e) {
            logger.error(e.getMessage(), e);
            returnMap.put("method", "sqlAccessFileExportResult");
            returnMap.put("resultCode", KeyValues.FAIL);
            returnMap.put("resultDesc", e.getMessage());
            SendMessageUtil.sendMessage(receiveClient, returnMap);
        }

    }

2.2hive或jdbc分布式数据集的获取

 /**
     * 获取数据集
     *
     * @param sparkSession sparkSession
     * @param paramsMap 参数
     * @return Dataset<Row>
     */
    private static Dataset<Row> getDataset(SparkSession sparkSession, Map<String, Object> paramsMap) {

        String datasourceType = MapParamUtil.getStringValue(paramsMap, "datasourceType");
        String url = MapParamUtil.getStringValue(paramsMap, "url");
        String username = MapParamUtil.getStringValue(paramsMap, "username");
        String password = MapParamUtil.getStringValue(paramsMap, "password");
        String driver = MapParamUtil.getStringValue(paramsMap, "driver");
        String sql = MapParamUtil.getStringValue(paramsMap, "sql");
        Long downloadTaskId = MapParamUtil.getLongValue(paramsMap, "downloadTaskId");

        if ("hive".equalsIgnoreCase(datasourceType)) {
            Dataset<Row> dataset = sparkSession.sql(sql);
            return dataset;
        }
        else {
            Dataset<Row> dataset = sparkSession.read().format("jdbc")
                // 地址
                .option("url", url)
                // 用户名
                .option("user", username)
                // 密码
                .option("password", new DesCryptUtil(DesCryptUtil.simpleKey).decrypt(password))
                // 驱动
                .option("driver", driver)
                // 生成临时表查询逻辑中,查询逻辑
                .option("dbtable", "(" + sql + ") as temp_" + downloadTaskId).load();
            return dataset;
        }
    }

2.3导出Excel代码示例


    /**
     * 创建excel文件
     *
     * @param dataset 数据集
     * @param fileLocation 文件路径
     * @param metaData 元数据翻译
     */
    private static void createExcel(Dataset<Row> dataset, String fileLocation, String taskName, String waterMark,
        Map<String, Object> metaData) {
        // 自动关闭流

        try (Workbook workbook = new XSSFWorkbook();
            FileOutputStream fileOutputStream = new FileOutputStream(fileLocation);) {

            CellStyle headStyle = workbook.createCellStyle();
            headStyle.setAlignment(HorizontalAlignment.CENTER);
            headStyle.setVerticalAlignment(VerticalAlignment.CENTER);
            headStyle.setFillPattern(FillPatternType.SOLID_FOREGROUND);
            headStyle.setFillForegroundColor(IndexedColors.GREY_25_PERCENT.getIndex());

            // 创建工作薄,写入列头
            Sheet sheet = workbook.createSheet(taskName);
            org.apache.poi.ss.usermodel.Row excelRow = sheet.createRow(0);

            String[] headersNames = dataset.columns();
            for (int i = 0; i < headersNames.length; i++) {
                Cell cell = excelRow.createCell(i);
                String headersName = headersNames[i];
                cell.setCellValue(MapParamUtil.getStringValue(metaData, headersName, headersName));
            }

            // 抽样数据到本地
            Iterator<Row> localIterator = dataset.toLocalIterator();
            while (localIterator.hasNext()) {
                Row row = localIterator.next();
                excelRow = sheet.createRow(sheet.getLastRowNum() + 1);
                for (int i = 0; i < headersNames.length; i++) {
                    Cell cell = excelRow.createCell(i);
                    cell.setCellValue(String.valueOf(row.get(i)));
                }
            }
            // 水印输出处理
            if (StringUtil.isNotEmpty(waterMark)) {
                ExcelWaterRemarkUtils.painWaterMarkToWorld((XSSFWorkbook) workbook, waterMark);
            }
            // 生成文件
            workbook.write(fileOutputStream);
            fileOutputStream.flush();
        }
        catch (Exception e) {
            logger.error(e.getMessage(), e);
            throw new SparkException(e.getMessage(), e);
        }
    }

2.4导出csv文件示例

   /**
     * 生成csv文件或者txt
     *
     * @param dataset 数据集
     * @param fileLocation 文件路径
     * @param metaData 元数据翻译
     */
    private static void createCsvOrTxt(Dataset<Row> dataset, String fileLocation, Map<String, Object> metaData) {
        try (FileOutputStream fileOutputStream = new FileOutputStream(fileLocation);
            OutputStreamWriter outputStreamWriter = new OutputStreamWriter(fileOutputStream, "UTF-8");
            BufferedWriter bufferedWriter = new BufferedWriter(outputStreamWriter);) {

            // 获取列头,进行元数据翻译操作
            String[] columns = dataset.columns();
            List<String> headersNames = new ArrayList<>(20);
            for (int i = 0; i < columns.length; i++) {
                String columnCode = columns[i];
                headersNames.add(MapParamUtil.getStringValue(metaData, columnCode, columnCode));
            }
            // 写入列头,用逗号分隔
            bufferedWriter.write(StringUtil.join(headersNames, ","));

            bufferedWriter.newLine();
            // 抽样数据到本地
            Iterator<Row> localIterator = dataset.toLocalIterator();
            while (localIterator.hasNext()) {
                Row row = localIterator.next();
                // 竖线分隔
                String mkString = row.mkString(",");
                bufferedWriter.write(mkString);
                bufferedWriter.newLine();
            }
            bufferedWriter.flush();
        }
        catch (Exception e) {
            logger.error(e.getMessage(), e);
            throw new SparkException(e.getMessage(), e);
        }
    }

2.5入paramMap入参参考

{
	"method": "sqlAccessFileExport",
	"appName": "xxx分析_develop_6924_2_20230420",
	"receiveClient": "http://xxx.xx.xx.xxx:8559/dataservice/",
	"url": "jdbc:postgresql://xxx.xx.xx.xx:5432/bigdata?gssEncMode\u003ddisable\u0026reWriteBatchedInserts\u003dtrue",
	"sql": "select product_name,charge_zy_sum,calculate979,product_name_count from smart_test.app_1681284022118 limit 100",
	"downloadTaskId": 42,
	"datasourceType": "gp",
	"tmpDir": "/home/xxxxx/xxxx/xxxxx/shuxkaiftp",
	"metaData": {
		"charge_zy_sum": "主营收入(万)",
		"product_name_count": "产品名称",
		"calculate979": "计算字段979",
		"product_name": "产品名称(产品维度)"
	},
	"password": "2047d192e697a909",
	"driver": "org.postgresql.Driver",
	"taskName": "xxx分析_develop_6924_2_20230420",
	"waterMark": "xxxx分析-(xxxuser) 2023-04-29 23:06:02",
	"fileType": "xlsx",
	"username": "gpxxxx"
}

到了这里,关于spark-sql(jdbc)本地模式导出csv或Excel文件的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • spark-sql

    [root@localhost bin]# ./spark-sql Error: Failed to load class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver. Failed to load main class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver. You need to build Spark with -Phive and -Phive-thriftserver. 24/02/22 00:23:20 INFO ShutdownHookManager: Shutdown hook called 24/02/22 00:23:20 INFO Shutd

    2024年02月22日
    浏览(41)
  • Spark-SQL小结

    目录 一、RDD、DataFrame、DataSet的概念、区别联系、相互转换操作   1.RDD概念   2.DataFrame概念   3.DataSet概念   4.RDD、DataFrame、DataSet的区别联系   5.RDD、DataFrame、DataSet的相互转换操作    1 RDD-DataFrame、DataSet    2  DataFrame-RDD,DataSet    3 DataSet-RDD,DataFrame 二、Spark-SQL连接JDBC的方式

    2024年02月09日
    浏览(44)
  • spark-sql字段血缘实现

    Apache Spark是一个开源的大数据处理框架,它提供了一种高效、易于使用的方式来处理大规模数据集。在Spark中,数据是通过DataFrame和Dataset的形式进行操作的,这些数据结构包含了一系列的字段(也称为列)。字段血缘是Spark中的一个关键概念,它帮助我们理解数据的来源和流

    2024年02月02日
    浏览(52)
  • Hudi-集成Spark之spark-sql方式

    启动spark-sql 创建表 建表参数: 参数名 默认值 说明 primaryKey uuid 表的主键名,多个字段用逗号分隔。同 hoodie.datasource.write.recordkey.field preCombineField 表的预合并字段。同 hoodie.datasource.write.precombine.field type cow 创建的表类型: type = ‘cow’ type = \\\'mor’同 hoodie.datasource.write.table.ty

    2024年02月05日
    浏览(46)
  • PL/SQL 中的数据导入和导出:CSV 文件格式详解

    前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站,这篇文章男女通用,看懂了就去分享给你的码吧。 在 PL/SQL 开发中,数据的导入和导出是常见的操作。本文将深入探讨如何使用 PL/SQL 导入和导出 CSV 文件格式的数据

    2024年02月12日
    浏览(60)
  • VUE通用后台管理系统(四)前端导出文件(CSV、XML、HTML、PDF、EXCEL)

    常见的导出格式:CSV、XML、HTML、PDF、EXCEL 1)准备工作 安装所需相关依赖 前两个是PDF格式需要的依赖,后两个是excel格式所需,如果没有需求这两种格式的可以忽略这一步 然后画页面   页面效果 2)导出CSV格式的文件 新建src/utils/utils.js文件 写入exportCsv方法,columns为表头,

    2024年02月05日
    浏览(57)
  • Spark参数配置和调优,Spark-SQL、Config

    一、Hive-SQL / Spark-SQL参数配置和调优 二、shell脚本spark-submit参数配置 三、sparkSession中配置参数

    2024年02月13日
    浏览(47)
  • Hudi(7):Hudi集成Spark之spark-sql方式

    目录 0. 相关文章链接 1. 创建表 1.1. 启动spark-sql 1.2. 建表参数 1.3. 创建非分区表 1.4. 创建分区表 1.5. 在已有的hudi表上创建新表 1.6. 通过CTAS (Create Table As Select)建表 2. 插入数据 2.1. 向非分区表插入数据 2.2. 向分区表动态分区插入数据 2.3. 向分区表静态分区插入数据 2.4

    2024年02月06日
    浏览(42)
  • Spark-SQL连接Hive的五种方法

    若使用Spark内嵌的Hive,直接使用即可,什么都不需要做(在实际生产活动中,很少会使用这一模式) 步骤: 将Hive中conf/下的hive-site.xml拷贝到Spark的conf/目录下; 把Mysql的驱动copy到jars/目录下; 如果访问不到hdfs,则将core-site.xml和hdfs-site.xml拷贝到conf/目录下; 重启spark-shell;

    2024年02月16日
    浏览(43)
  • spark-sql: insert overwrite分区表问题

    用spark-sql,insert overwrite分区表时发现两个比较麻烦的问题: 从目标表select出来再insert overwrite目标表时报错:Error in query: Cannot overwrite a path that is also being read from. 从其他表select出来再insert overwrite目标表时,其他分区都被删除了. 印象中这两个问题也出现过,但凭经验和感觉,

    2024年02月11日
    浏览(45)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包