mysql数据快速导入doris

背景

前段时间业务需要将mysql数据导入到doris ，以便大数据平台使用

问题

本来想法很简单，doris 语法兼容mysql,将数据导出为insert 语句，直接插入就行。
想法不错，但是奈何数据量大（200多w），插入几个钟头还没完事。后来想想，试试批量insert语句。也挺慢。听说csv导入doris挺快的，奈何又遇到分隔符问题，简单说就是没有转义，导致列数不对。难道没有完美方法

解决

办法还是有的，可以将插入sql转换为csv，并转义（分界符），代码如下：

importjava.io.BufferedReader;importjava.io.FileReader;importjava.io.FileWriter;importjava.io.IOException;importjava.util.ArrayList;importjava.util.List;/***简化但可靠的SQL转CSV工具*/publicclassInsertSqlToCSV{publicstaticvoidmain(String[]args)throwsIOException{if(args.length<2){System.out.println("使用方法: java InsertSqlToCSV <输入SQL文件> <输出CSV文件>");return;}convertSqlToCsv(args[0],args[1]);}publicstaticvoidconvertSqlToCsv(StringsqlFile,StringcsvFile)throwsIOException{List<String[]>allData=newArrayList<>();List<String>headers=null;inttotalRows=0;intsuccessRows=0;interrorRows=0;System.out.println("开始处理: "+sqlFile);try(BufferedReaderbr=newBufferedReader(newFileReader(sqlFile));FileWriterfw=newFileWriter(csvFile)){Stringline;intlineNum=0;while((line=br.readLine())!=null){lineNum++;line=line.trim();if(line.toUpperCase().startsWith("INSERT INTO")){totalRows++;try{//提取字段名（只从第一行提取）if(headers==null){headers=extractHeaders(line);System.out.println("提取到 "+headers.size()+" 个字段");}//提取VALUES部分StringvaluesPart=extractValuesPart(line);//解析VALUESList<String>values=parseSimple(valuesPart);//验证列数if(values.size()!=headers.size()){System.out.println("第 "+lineNum+" 行: 期望 "+headers.size()+" 列，实际 "+values.size()+" 列");errorRows++;continue;}allData.add(values.toArray(newString[0]));successRows++;}catch(Exceptione){System.out.println("第 "+lineNum+" 行解析失败: "+e.getMessage());errorRows++;}}}System.out.println("\n处理完成:");System.out.println("总INSERT语句: "+totalRows);System.out.println("成功: "+successRows);System.out.println("失败: "+errorRows);if(successRows==0){System.out.println("没有成功解析的数据！");return;}//写入CSVwriteCsv(fw,headers,allData);System.out.println("CSV文件已生成: "+csvFile);}catch(Exceptione){System.err.println("处理失败: "+e.getMessage());e.printStackTrace();}}/***提取字段名*/privatestaticList<String>extractHeaders(Stringsql){List<String>headers=newArrayList<>();//找到字段列表的开始intstart=sql.indexOf('(');if(start==-1)returnheaders;//找到字段列表的结束intend=sql.indexOf(')',start);if(end==-1)returnheaders;StringheaderPart=sql.substring(start+1,end).trim();//简单分割字段名String[]parts=headerPart.split(",");for(Stringpart:parts){Stringheader=part.trim();//移除引号if(header.startsWith("`")&&header.endsWith("`")){header=header.substring(1,header.length()-1);}elseif(header.startsWith("\"")&&header.endsWith("\"")){header=header.substring(1,header.length()-1);}elseif(header.startsWith("'")&&header.endsWith("'")){header=header.substring(1,header.length()-1);}elseif(header.startsWith("[")&&header.endsWith("]")){header=header.substring(1,header.length()-1);}headers.add(header);}returnheaders;}/***提取VALUES部分*/privatestaticStringextractValuesPart(Stringsql){//找到VALUES关键字intvaluesIndex=sql.toUpperCase().indexOf("VALUES");if(valuesIndex==-1){thrownewIllegalArgumentException("未找到VALUES关键字");}StringafterValues=sql.substring(valuesIndex+"VALUES".length()).trim();//移除末尾分号if(afterValues.endsWith(";")){afterValues=afterValues.substring(0,afterValues.length()-1);}//找到括号内容intopenParen=afterValues.indexOf('(');intcloseParen=afterValues.lastIndexOf(')');if(openParen==-1||closeParen==-1||closeParen<=openParen){thrownewIllegalArgumentException("未找到有效的VALUES括号");}returnafterValues.substring(openParen+1,closeParen).trim();}/***简单但可靠的解析器*/privatestaticList<String>parseSimple(StringvaluesStr){List<String>values=newArrayList<>();StringBuildercurrent=newStringBuilder();booleaninQuotes=false;charquoteChar='\0';for(inti=0;i<valuesStr.length();i++){charc=valuesStr.charAt(i);if(!inQuotes){//不在引号内if(c=='\''||c=='"'){inQuotes=true;quoteChar=c;current.append(c);}elseif(c==','){//字段分隔符Stringvalue=cleanValue(current.toString().trim());values.add(value);current.setLength(0);}else{current.append(c);}}else{//在引号内current.append(c);if(c==quoteChar){//检查是否是转义引号if(i+1<valuesStr.length()&&valuesStr.charAt(i+1)==quoteChar){//是转义引号，跳过下一个current.append(quoteChar);i++;}else{//是结束引号inQuotes=false;}}}}//最后一个值if(current.length()>0){Stringvalue=cleanValue(current.toString().trim());values.add(value);}returnvalues;}/***清理值*/privatestaticStringcleanValue(Stringvalue){value=value.trim();if(value.equalsIgnoreCase("NULL")){return"";}//移除外层引号if((value.startsWith("'")&&value.endsWith("'"))||(value.startsWith("\"")&&value.endsWith("\""))){value=value.substring(1,value.length()-1);//处理转义引号value=value.replace("''","'").replace("\"\"","\"");}returnvalue;}/***写入CSV*/privatestaticvoidwriteCsv(FileWriterwriter,List<String>headers,List<String[]>data)throwsIOException{//写入表头List<String>escapedHeaders=newArrayList<>();for(Stringheader:headers){escapedHeaders.add(escapeCsv(header));}writer.write(String.join(",",escapedHeaders));writer.write("\n");//写入数据for(String[]row:data){List<String>escapedRow=newArrayList<>();for(Stringcell:row){escapedRow.add(escapeCsv(cell));}writer.write(String.join(",",escapedRow));writer.write("\n");}}/***CSV转义*/privatestaticStringescapeCsv(Stringvalue){if(value==null)return"";//if(value.contains(",")||value.contains("\"")||//value.contains("\n")||value.contains("\r")){//value=value.replace("\"","\"\"");//return"\""+value+"\"";//}//移除可能已经存在的外层单引号Stringprocessed=value.trim();booleanhadQuotes=false;if(processed.startsWith("'")&&processed.endsWith("'")&&processed.length()>1){processed=processed.substring(1,processed.length()-1);hadQuotes=true;}//转义处理StringBuilderresult=newStringBuilder();for(inti=0;i<processed.length();i++){charc=processed.charAt(i);if(c=='\\'){//反斜杠转义result.append("\\\\");}elseif(c=='\''){//单引号转义result.append("\\'");}else{result.append(c);}}//如果原始值有引号或者我们需要加引号，就加上if(hadQuotes||!processed.isEmpty()){return"'"+result.toString()+"'";}else{returnresult.toString();}}}

这样处理后，250万行的sql转csv，大概10分钟左右，然后csv导入doris ，几秒就行

最后

如果那位有更好办法，可以联系我：lita2lz