Apps Script로 Google Sheets의 데이터를 BigQuery에 Load

💡Apps Script에서 Google Sheets에 입력한 데이터를 BigQuery에 저장하는 방법을 알아본다.

 

이제 Google Sheets의 데이터를 BigQuery에 저장해본다.

BigQuery Service | Apps Script | Google Developers

BigQuery 서비스  |  Apps Script  |  Google Developers

데이터를 입력할 sheet와 table을 준비한다.

1. 기본 저장 로직

insert 정보 설정 및 실행

function insertSheetData() {
    const sheetUrl  = SpreadsheetApp.getActiveSpreadsheet().getUrl();
    const sheetName = "Sheet3"
    const sheet     = SpreadsheetApp.openByUrl(sheetUrl).getSheetByName(sheetName);
    // insert table
    const table_id  = "insert_test";
    // sheet에서 저장할 영역부
    const colInfo = {startRow:2, startColumn:1};
    // 실행부
    const resultFlag = loadSpreadsheet(sheet, global_datasetId, table_id, colInfo);

    console.log(`Result Flag : ${resultFlag}`);
}

/**
 * Load Spredsheet to Bigquery
 * @param {object} sheet        = 작업할 sheet
 * @param {string} datasetId    = 데이터셋 명
 * @param {string} tableId      = 테이블 명
 * @param {object} colInfo      = 데이터가공을 위한 컬럼정보
 */
function loadSpreadsheet(sheet, datasetId, tableId, colInfo) {
  var projectId = GLOBAL_PROJECT_ID;
  var errMsg = "";

  var writeDispositionSetting = 'WRITE_APPEND';
 
  var lastColumn  = sheet.getDataRange().getLastColumn();
  var getLastRow  = sheet.getDataRange().getLastRow();
  var endRow      = (parseInt(getLastRow) - parseInt(colInfo.startRow) + 1);
  // getSheetValues를 이용해 영역을 지정하여 값을 가져옴
  var rows        = sheet.getSheetValues( colInfo.startRow
																				, colInfo.startColumn
																				, endRow
																				, lastColumn);
  var rowsCSV     = rows.join("\\n");
  var blob        = Utilities.newBlob(rowsCSV, "text/csv");
  var data        = blob.setContentType('application/octet-stream');

  // Create the data upload job.
  var job = {
    configuration: {
      load: {
        destinationTable: {
          projectId: projectId,
          datasetId: datasetId,
          tableId: tableId,
          location: GLOBAL_LOCATION
        },
        writeDisposition: writeDispositionSetting
      }
    },
    jobReference: {
        projectId: projectId,
        useLegacySql: false,
        location: GLOBAL_LOCATION
    }, 
    jobTimeoutMs:100000
  };
  
  // send the job to BigQuery so it will run your query
  var runJob = BigQuery.Jobs.insert(job, projectId, data);
  var jobId = runJob.jobReference.jobId
  
  var status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});

  var flag = false;

  // 돌자마자 에러메세지 발생 case
  if ( status.status.state != "RUNNING" ) {
    if ( status.status.errors ) {
      Logger.log('Error: ' + status.status.errors);
      errMsg += status.status.errors[0].message + "\\r\\n";
      errMsg += status.status.errors[1].message;
      flag = false;
    } else {
      flag = true;
    }
  }

  while (status.status.state === 'RUNNING') {
    Utilities.sleep(100);
    status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});
    if ( status.status.errors ) {
      Logger.log('Error: ' + status.status.errors);
      errMsg += status.status.errors[0].message + "\\r\\n";
      errMsg += status.status.errors[1].message;
      flag = false;
    } else {
      flag = true;
    }
  }

  if ( flag ) {
    Logger.log('FINNISHED!');
  } else {
    Logger.log('Failed! ' + errMsg);
  }
  return flag;
}

정상적으로 저장된 것을 확인할 수 있다.

 

 

2. Date 타입 저장

생년월일 (DATE 타입) 도 추가해본다.

 

 

 

저장해보니 에러가 발생한다.

Error while reading data, error message: Could not parse 
'Sat Oct 11 1980 00:00:00 GMT+0900 (Korean Standard Time)' as DATE for field birth 
(position 4) starting at location 0  with message 'Unable to parse'"

Google Sheets에서 Date type이 다르게 나타난다.

Sat Oct 11 1980 00:00:00 GMT+0900 (Korean Standard Time) => 1980-10-11 Sun Feb 11 1990 00:00:00 GMT+0900 (Korean Standard Time) => 1990-02-11 Mon Aug 01 1988 01:00:00 GMT+1000 (Korean Daylight Time) => 1988-08-01

변경하는 로직을 추가해야한다.

Sheet 데이터를 가져와 date 타입을 변경하는 convertDateFormat 라는 로직을 추가한다.

 

 

 

/**
 * Load Spredsheet to Bigquery
 * @param {object} sheet        = 작업할 sheet
 * @param {string} datasetId    = 데이터셋 명
 * @param {string} tableId      = 테이블 명
 * @param {object} colInfo      = 데이터가공을 위한 컬럼정보
 */
function loadSpreadsheet(sheet, datasetId, tableId, colInfo) {
    var projectId = GLOBAL_PROJECT_ID;
    var errMsg = "";

    var writeDispositionSetting = 'WRITE_APPEND';

    var lastColumn  = sheet.getDataRange().getLastColumn();
    var getLastRow  = sheet.getDataRange().getLastRow();
    var endRow      = (parseInt(getLastRow) - parseInt(colInfo.startRow) + 1);
    // getSheetValues를 이용해 영역을 지정하여 값을 가져옴
    var orgin_rows  = sheet.getSheetValues( colInfo.startRow
                                            , colInfo.startColumn
                                            , endRow
                                            , lastColumn);
    var rows        = convertDataStructure(orgin_rows);
    var rowsCSV     = rows.join("\n");
    var blob        = Utilities.newBlob(rowsCSV, "text/csv");
    var data        = blob.setContentType('application/octet-stream');

    // Create the data upload job.
    var job = {
        configuration: {
            load: {
                destinationTable: {
                    projectId: projectId,
                    datasetId: datasetId,
                    tableId: tableId,
                    location: GLOBAL_LOCATION
                },
            	writeDisposition: writeDispositionSetting
            }
        },
        jobReference: {
            projectId: projectId,
            useLegacySql: false,
            location: GLOBAL_LOCATION
        }, 
        jobTimeoutMs:100000
    };

    // send the job to BigQuery so it will run your query
    var runJob = BigQuery.Jobs.insert(job, projectId, data);
    var jobId  = runJob.jobReference.jobId

    var status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});

    var flag = false;

    // 돌자마자 에러메세지 발생 case
    if ( status.status.state != "RUNNING" ) {
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\r\n";
            errMsg += status.status.errors[1].message;
            flag = false;
        } else {
        	flag = true;
        }
    }

    while (status.status.state === 'RUNNING') {
        Utilities.sleep(100);
        status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\r\n";
            errMsg += status.status.errors[1].message;
            flag = false;
        } else {
        	flag = true;
        }
    }

    if ( flag ) {
    	Logger.log('FINNISHED!');
    } else {
    	Logger.log('Failed! ' + errMsg);
    }
    return flag;
}



/** 
 * Spreadsheet 데이터를 Bigquery에 저장하기위해 데이터 형식 변환
 * @param {object}  rows  = Spredsheet 데이터
 */
function convertDataStructure(rows) {
  var returnObj = new Array();
  if (rows) {
    for (var i = 0; i < rows.length; i++) {
      var cols = rows[i];
      cols[4] = convertDateFormat(cols[4], "-");
      returnObj.push(cols);
    }
  }
  return returnObj;
}

/**
 * Data string to Data format.
 * @param {string}  dateString   = 원 문자열
 * @param {string}  format       = 연결 문자
 */
function convertDateFormat(dateString, format) {
  var convertDate = new Date(dateString);
  if ( isNull(dateString) ) {
    return "";
  }
  if ( isNull(format) ) {
    format = "";
  }
  var convertString = convertDate.getFullYear() + format + lpad((convertDate.getMonth() +1), 2, "0") + format + lpad(convertDate.getDate(), 2, "0");
  return convertString;
}

/**
 * Null 체크
 * @param {obj}  obj  = Null 체크 대상
 */
function isNull(obj) {
	return (typeof obj != "undefined" && obj != null && obj != "") ? false : true;
}

/**
 * 좌측 문자열 채우기
 * @param {string}  str     = 원 문자열
 * @param {int}     padLen  = 최대 채우고자 하는 길이
 * @param {string}  padStr  = 채우고자하는 문자(char)
 */
function lpad(str, padLen, padStr) {
    if ( padStr.length > padLen ) {
      console.log("오류 : 채우고자 하는 문자열이 요청 길이보다 큽니다");
      return str;
    }
    str    += ""; // 문자로
    padStr += ""; // 문자로
    while (str.length < padLen) {
      str = padStr + str;
    }
    str = str.length >= padLen ? str.substring(0, padLen) : str;
    return str;
}

데이터가 정상적으로 저장되었다.

 

 

3. Google Sheets 에 저장되지 않는 기초데이터 함께 저장하기

데이터를 저장할 때, sheet에 없는 데이터도 함께 저장하고싶다.

status 값, savetime 등…

addValues 를 만들어서 저장하는 함수에 함께 넘겨준다.

convertStructure 함수에서 뒤에 값들을 붙여준다.

function insertSheetData() {
  const sheetUrl  = SpreadsheetApp.getActiveSpreadsheet().getUrl();
  const sheetName = "Sheet3"
  const sheet     = SpreadsheetApp.openByUrl(sheetUrl).getSheetByName(sheetName);
  const table_id  = "insert_test";

  const colInfo = {startRow:2, startColumn:1};

  const saveTime = getDateTime("BQ");
  const addValues = {
                        status      : "Y"
                        , saveTime  : saveTime
                    };

  const resultFlag = loadSpreadsheet(sheet, global_datasetId, table_id, colInfo, addValues);

  console.log(`Result Flag : ${resultFlag}`);

}

/**
 * Load Spredsheet to Bigquery
 * @param {object} sheet        = 작업할 sheet
 * @param {string} datasetId    = 데이터셋 명
 * @param {string} tableId      = 테이블 명
 * @param {object} colInfo      = 데이터가공을 위한 컬럼정보
 * @param {object} addValues    = 가공할 데이터 정보들
 */
function loadSpreadsheet(sheet, datasetId, tableId, colInfo, addValues) {
    var projectId = GLOBAL_PROJECT_ID;
    var errMsg = "";

    var writeDispositionSetting = 'WRITE_APPEND';

    var lastColumn  = sheet.getDataRange().getLastColumn();
    var getLastRow  = sheet.getDataRange().getLastRow();
    var endRow      = (parseInt(getLastRow) - parseInt(colInfo.startRow) + 1);
    // getSheetValues를 이용해 영역을 지정하여 값을 가져옴
    var orgin_rows  = sheet.getSheetValues( colInfo.startRow
                                            , colInfo.startColumn
                                            , endRow
                                            , lastColumn);
    var rows        = convertDataStructure(orgin_rows, addValues);
    var rowsCSV     = rows.join("\n");
    var blob        = Utilities.newBlob(rowsCSV, "text/csv");
    var data        = blob.setContentType('application/octet-stream');

    // Create the data upload job.
    var job = {
        configuration: {
            load: {
                destinationTable: {
                    projectId: projectId,
                    datasetId: datasetId,
                    tableId: tableId,
                    location: GLOBAL_LOCATION
                },
            writeDisposition: writeDispositionSetting
            }
        },
        jobReference: {
            projectId: projectId,
            useLegacySql: false,
            location: GLOBAL_LOCATION
        }, 
        jobTimeoutMs:100000
    };

    // send the job to BigQuery so it will run your query
    var runJob = BigQuery.Jobs.insert(job, projectId, data);
    var jobId  = runJob.jobReference.jobId

    var status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});

    var flag = false;

    // 돌자마자 에러메세지 발생 case
    if ( status.status.state != "RUNNING" ) {
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\r\n";
            errMsg += status.status.errors[1].message;
            flag = false;
        } else {
        	flag = true;
        }
    }

    while (status.status.state === 'RUNNING') {
        Utilities.sleep(100);
        status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\r\n";
            errMsg += status.status.errors[1].message;
            flag = false;
        } else {
        	flag = true;
        }
    }

    if ( flag ) {
    	Logger.log('FINNISHED!');
    } else {
    	Logger.log('Failed! ' + errMsg);
    }
    return flag;
}



/** 
 * Spreadsheet 데이터를 Bigquery에 저장하기위해 데이터 형식 변환
 * @param {object}  rows  = Spredsheet 데이터
 */
function convertDataStructure(rows) {
  var returnObj = new Array();
  if (rows) {
    for (var i = 0; i < rows.length; i++) {
      var cols = rows[i];
      cols[4] = convertDateFormat(cols[4], "-");

			if ( !isNull(addValues) ) {
        for ( key in addValues ) {
          cols.push(addValues[key]); 
        }
      }

      returnObj.push(cols);
    }
  }
  return returnObj;
}

/**
 * Null 체크
 * @param {obj}  obj  = Null 체크 대상
 */
function isNull(obj) {
	return (typeof obj != "undefined" && obj != null && obj != "") ? false : true;
}

/**
 * 좌측 문자열 채우기
 * @param {string}  str     = 원 문자열
 * @param {int}     padLen  = 최대 채우고자 하는 길이
 * @param {string}  padStr  = 채우고자하는 문자(char)
 */
function lpad(str, padLen, padStr) {
    if ( padStr.length > padLen ) {
      console.log("오류 : 채우고자 하는 문자열이 요청 길이보다 큽니다");
      return str;
    }
    str    += ""; // 문자로
    padStr += ""; // 문자로
    while (str.length < padLen) {
      str = padStr + str;
    }
    str = str.length >= padLen ? str.substring(0, padLen) : str;
    return str;
}

Apps Script에서 BigQuery에 저장하기 위한 현재시간 Datetime 변수생성

/**
 * Data string to Data format.
 * @param {string}  dateString   = 원 문자열
 * @param {string}  format       = 연결 문자
 */
function convertDateFormat(dateString, format) {
  var convertDate = new Date(dateString);
  if ( isNull(dateString) ) {
    return "";
  }
  if ( isNull(format) ) {
    format = "";
  }
  var convertString = convertDate.getFullYear() + format + lpad((convertDate.getMonth() +1), 2, "0") + format + lpad(convertDate.getDate(), 2, "0");
  return convertString;
}

 

 

4. 셀 값 중 콤마나 더블쿼테이션 처리

이대로 저장하게 되면 에러가 발생한다.

데이터 저장시 sheet 데이터를 csv형태로 변경하여 처리하기 때문에

데이터에 콤마가 들어가 있으면 에러가 발생한다.

테이블에는 8개의 컬럼이 있지만, 9개의 컬럼을 저장하려고 한다는 에러.

(더블쿼테이션은 생략된다)

Failed! Error while reading data, error message: 
Too many values in row starting at position: 74. Found 9 column(s) while expected 8.
Error while reading data, error message: CSV processing encountered too many errors

 

 

저장로직에서 아래 rowCsv 영역을 수정하여 저장하면,

더블쿼테이션과 따옴표의 정상적인 저장처리가 가능하다.

/**
 * Load Spredsheet to Bigquery
 * @param {object} sheet        = 작업할 sheet
 * @param {string} datasetId    = 데이터셋 명
 * @param {string} tableId      = 테이블 명
 * @param {object} colInfo      = 데이터가공을 위한 컬럼정보
 * @param {object} addValues    = 가공할 데이터 정보들
 */
function loadSpreadsheet(sheet, datasetId, tableId, colInfo, addValues) {
    var projectId = GLOBAL_PROJECT_ID;
    var errMsg = "";

    var writeDispositionSetting = 'WRITE_APPEND';

    var lastColumn  = sheet.getDataRange().getLastColumn();
    var getLastRow  = sheet.getDataRange().getLastRow();
    var endRow      = (parseInt(getLastRow) - parseInt(colInfo.startRow) + 1);
    // getSheetValues를 이용해 영역을 지정하여 값을 가져옴
    var orgin_rows  = sheet.getSheetValues( colInfo.startRow
                                            , colInfo.startColumn
                                            , endRow
                                            , lastColumn);
    var rows        = convertDataStructure(orgin_rows, addValues);

    //  var rowsCSV     = rows.join("\\n");
    //  var blob        = Utilities.newBlob(rowsCSV, "text/csv");
    //  var data        = blob.setContentType('application/octet-stream');

    // 기존 csv 형태는 콤마가 안되서 변경한 부분
    var rowsCSV = rows.map(values => values.map
                                    (value => JSON.stringify(value).replace(/\\\\"/g, '""'))
                            );
    var csvData = rowsCSV.map(values => values.join(',')).join('\\n');
    var data = Utilities.newBlob(csvData, 'application/octet-stream');

    // Create the data upload job.
    var job = {
        configuration: {
            load: {
                destinationTable: {
                    projectId: projectId,
                    datasetId: datasetId,
                    tableId: tableId,
                    location: GLOBAL_LOCATION
                },
                writeDisposition: writeDispositionSetting
            }
        },
        jobReference: {
            projectId: projectId,
            useLegacySql: false,
            location: GLOBAL_LOCATION
        }, 
        jobTimeoutMs:100000
    };

    // send the job to BigQuery so it will run your query
    var runJob = BigQuery.Jobs.insert(job, projectId, data);
    var jobId  = runJob.jobReference.jobId

    var status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});

    var flag = false;

    // 돌자마자 에러메세지 발생 case
    if ( status.status.state != "RUNNING" ) {
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\\r\\n";
            errMsg += status.status.errors[1].message;
            flag = false;
        } else {
        	flag = true;
        }
    }

    while (status.status.state === 'RUNNING') {
        Utilities.sleep(100);
        status = BigQuery.Jobs.get(projectId, jobId, {location: GLOBAL_LOCATION});
        if ( status.status.errors ) {
            Logger.log('Error: ' + status.status.errors);
            errMsg += status.status.errors[0].message + "\\r\\n";
            errMsg += status.status.errors[1].message;
        	flag = false;
        } else {
        	flag = true;
        }
    }

    if ( flag ) {
    	Logger.log('FINNISHED!');
    } else {
    	Logger.log('Failed! ' + errMsg);
    }
    return flag;
}