喜当爹

Posted on 2024-10-27

意不意外、惊不惊喜，就是这么快。

自定义图床

Posted on 2024-10-27

之前用的阿里云图床，后来产生的费用让人难以维持，下面用TG+CF的方案实现一个图床，自己试了一下很好用

https://img.yihao.de/

一、GitHub

按照github的代码部署一下，操作如readme一样

0-RTT/telegraph

二、 worder.js

我这边自行稍微改造了一点

export default {
  async fetch(request, env) {
    const { pathname } = new URL(request.url);
    const domain = env.DOMAIN;
    const DATABASE = env.DATABASE;
    const USERNAME = env.USERNAME;
    const PASSWORD = env.PASSWORD;
    const adminPath = env.ADMIN_PATH;
    const enableAuth = env.ENABLE_AUTH === 'true';
    const TG_BOT_TOKEN = env.TG_BOT_TOKEN;
    const TG_CHAT_ID = env.TG_CHAT_ID;

    switch (pathname) {
      case '/':
        return await handleRootRequest(request, USERNAME, PASSWORD, enableAuth);
      case `/${adminPath}`:
        return await handleAdminRequest(DATABASE, request, USERNAME, PASSWORD);
      case '/upload':
        return request.method === 'POST' ? await handleUploadRequest(request, DATABASE, enableAuth, USERNAME, PASSWORD, domain, TG_BOT_TOKEN, TG_CHAT_ID) : new Response('Method Not Allowed', { status: 405 });
      case '/bing-images':
        return handleBingImagesRequest();
      case '/delete-images':
        return handleDeleteImagesRequest(request, DATABASE);
      default:
        return await handleImageRequest(request, DATABASE, TG_BOT_TOKEN);
    }
  }
};

let isAuthenticated = false;

function authenticate(request, USERNAME, PASSWORD) {
  const authHeader = request.headers.get('Authorization');
  if (!authHeader) return false;
  return isValidCredentials(authHeader, USERNAME, PASSWORD);
}

async function handleRootRequest(request, USERNAME, PASSWORD, enableAuth) {
  const cache = caches.default;
  const cacheKey = new Request(request.url);
  if (enableAuth) {
      if (!authenticate(request, USERNAME, PASSWORD)) {
          return new Response('Unauthorized', { status: 401, headers: { 'WWW-Authenticate': 'Basic realm="Admin"' } });
      }
  }
  const cachedResponse = await cache.match(cacheKey);
  if (cachedResponse) {
      return cachedResponse;
  }
  const response = new Response(`
  <!DOCTYPE html>
  <html lang="zh-CN">
  <head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <meta name="description" content="Telegraph图床-基于Workers的图床服务">
  <meta name="keywords" content="Telegraph图床,Workers图床, Cloudflare, Workers,telegra.ph, 图床">
  <title>Telegraph图床-基于Workers的图床服务</title>
  <link rel="icon" href="https://p1.meituan.net/csc/c195ee91001e783f39f41ffffbbcbd484286.ico" type="image/x-icon">
  <link href="https://lf3-cdn-tos.bytecdntp.com/cdn/expire-1-M/twitter-bootstrap/4.6.1/css/bootstrap.min.css" rel="stylesheet" />
  <link href="https://lf26-cdn-tos.bytecdntp.com/cdn/expire-1-M/bootstrap-fileinput/5.2.7/css/fileinput.min.css" rel="stylesheet" />
  <link href="https://lf26-cdn-tos.bytecdntp.com/cdn/expire-1-M/toastr.js/2.1.4/toastr.min.css" rel="stylesheet" />
  <link href="https://lf3-cdn-tos.bytecdntp.com/cdn/expire-1-M/font-awesome/5.15.4/css/all.min.css" type="text/css" rel="stylesheet" />
  <style>
      body {
          margin: 0;
          display: flex;
          justify-content: center;
          align-items: center;
          height: 100vh;
          position: relative;
      }
      .background {
          position: absolute;
          top: 0;
          left: 0;
          width: 100%;
          height: 100%;
          background-size: cover;
          z-index: -1;
          transition: opacity 1s ease-in-out;
          opacity: 1;
      }
      .card {
          background-color: rgba(255, 255, 255, 0.8);
          border: none;
          border-radius: 10px;
          box-shadow: 0 4px 8px rgba(0, 0, 0, 0.2);
          padding: 20px;
          width: 90%;
          max-width: 400px;
          text-align: center;
          margin: 0 auto;
          position: relative;
      }
      .uniform-height {
          margin-top: 20px;
      }
      #viewCacheBtn {
          position: absolute;
          top: 10px;
          right: 10px;
          background: none;
          border: none;
          color: rgba(0, 0, 0, 0.1);
          cursor: pointer;
          font-size: 24px;
          transition: color 0.3s ease;
      }
      #viewCacheBtn:hover {
          color: rgba(0, 0, 0, 0.4);
      }
      #cacheContent {
          margin-top: 20px;
          max-height: 200px;
          border-radius: 5px;
          overflow-y: auto;
      }
      .cache-title {
          text-align: left;
          margin-bottom: 10px;
      }
      .cache-item {
          display: block;
          cursor: pointer;
          border-radius: 4px;
          box-shadow: 0 1px 3px rgba(0, 0, 0, 0.1);
          transition: background-color 0.3s ease;
          text-align: left;
          padding: 10px;
      }
      .cache-item:hover {
          background-color: #e9ecef;
      }
      .project-link {
          font-size: 14px;
          text-align: center;
          margin-top: 5px;
          margin-bottom: 0;
      }
      textarea.form-control {
          max-height: 200px;
          overflow-y: hidden;
          resize: none;
      }
  </style>
</head>
<body>
  <div class="background" id="background"></div>
  <div class="card">
      <div class="title">Telegraph图床</div>
      <button type="button" class="btn" id="viewCacheBtn" title="查看历史记录"><i class="fas fa-clock"></i></button>
      <div class="card-body">
          <form id="uploadForm" action="/upload" method="post" enctype="multipart/form-data">
              <div class="file-input-container">
                  <input id="fileInput" name="file" type="file" class="form-control-file" data-browse-on-zone-click="true" multiple>
              </div>
              <div class="form-group mb-3 uniform-height" style="display: none;">
                  <button type="button" class="btn btn-light mr-2" id="urlBtn">URL</button>
                  <button type="button" class="btn btn-light mr-2" id="bbcodeBtn">BBCode</button>
                  <button type="button" class="btn btn-light" id="markdownBtn">Markdown</button>
              </div>
              <div class="form-group mb-3 uniform-height" style="display: none;">
                  <textarea class="form-control" id="fileLink" readonly></textarea>
              </div>
              <div id="cacheContent" style="display: none;"></div>
          </form>
      </div>
      <script src="https://lf3-cdn-tos.bytecdntp.com/cdn/expire-1-M/jquery/3.6.0/jquery.min.js" type="application/javascript"></script>
      <script src="https://lf26-cdn-tos.bytecdntp.com/cdn/expire-1-M/bootstrap-fileinput/5.2.7/js/fileinput.min.js" type="application/javascript"></script>
      <script src="https://lf26-cdn-tos.bytecdntp.com/cdn/expire-1-M/bootstrap-fileinput/5.2.7/js/locales/zh.min.js" type="application/javascript"></script>
      <script src="https://lf9-cdn-tos.bytecdntp.com/cdn/expire-1-M/toastr.js/2.1.4/toastr.min.js" type="application/javascript"></script>
      <script>
          async function fetchBingImages() {
              const response = await fetch('/bing-images');
              const data = await response.json();
              return data.data.map(image => image.url);
          }

          async function setBackgroundImages() {
              const images = await fetchBingImages();
              const backgroundDiv = document.getElementById('background');
              if (images.length > 0) {
                  backgroundDiv.style.backgroundImage = 'url(' + images[0] + ')';
              }
              let index = 0;
              let currentBackgroundDiv = backgroundDiv;
              setInterval(() => {
                  const nextIndex = (index + 1) % images.length;
                  const nextBackgroundDiv = document.createElement('div');
                  nextBackgroundDiv.className = 'background next';
                  nextBackgroundDiv.style.backgroundImage = 'url(' + images[nextIndex] + ')';
                  document.body.appendChild(nextBackgroundDiv);
                  nextBackgroundDiv.style.opacity = 0;
                  setTimeout(() => {
                      nextBackgroundDiv.style.opacity = 1;
                  }, 50);
                  setTimeout(() => {
                      document.body.removeChild(currentBackgroundDiv);
                      currentBackgroundDiv = nextBackgroundDiv;
                      index = nextIndex;
                  }, 1000);
              }, 5000);
          }

          $(document).ready(function() {
              let originalImageURLs = [];
              let isCacheVisible = false;
              initFileInput();
              setBackgroundImages();

              function initFileInput() {
                  $("#fileInput").fileinput({
                      theme: 'fa',
                      language: 'zh',
                      browseClass: "btn btn-primary",
                      removeClass: "btn btn-danger",
                      showUpload: false,
                      showPreview: false,
                  }).on('filebatchselected', handleFileSelection)
                    .on('fileclear', handleFileClear);
              }

              async function handleFileSelection() {
                  const files = $('#fileInput')[0].files;
                  for (let i = 0; i < files.length; i++) {
                      await uploadFile(files[i]);
                  }
              }

              async function uploadFile(file) {
                  try {
                      toastr.info('上传中...', '', { timeOut: 0 });
                      const interfaceInfo = {
                          acceptTypes: 'image/*,video/*',
                          maxFileSize: 20 * 1024 * 1024
                      };
                      const acceptedTypes = interfaceInfo.acceptTypes.split(',');
                      const isAcceptedType = acceptedTypes.some(type => {
                          return type.includes('*') ? file.type.startsWith(type.split('/')[0]) : file.type === type;
                      });
                      if (!isAcceptedType) {
                          toastr.error('仅支持图片或视频格式的文件。');
                          return;
                      }
                      if (file.size > interfaceInfo.maxFileSize) {
                          if (file.type.startsWith('video/') || file.type === 'image/gif') {
                              toastr.error('文件必须≤20MB');
                              return;
                          } else {
                              toastr.info('正在压缩...', '', { timeOut: 0 });
                              const compressedFile = await compressImage(file);
                              file = compressedFile;
                          }
                      }
                      const formData = new FormData($('#uploadForm')[0]);
                      formData.set('file', file, file.name);
                      const uploadResponse = await fetch('/upload', { method: 'POST', body: formData });
                      const responseData = await handleUploadResponse(uploadResponse);
                      if (responseData.error) {
                          toastr.error(responseData.error);
                      } else {
                          originalImageURLs.push(responseData.data);
                          $('#fileLink').val(originalImageURLs.join('\\n\\n'));
                          $('.form-group').show();
                          adjustTextareaHeight($('#fileLink')[0]);
                          toastr.success('文件上传成功！');
                          saveToLocalCache(responseData.data, file.name);
                      }
                  } catch (error) {
                      console.error('处理文件时出现错误:', error);
                      $('#fileLink').val('文件处理失败！');
                      toastr.error('文件处理失败！');
                  } finally {
                      toastr.clear();
                  }
              }

              async function handleUploadResponse(response) {
                  if (response.ok) {
                      return await response.json();
                  } else {
                      const errorData = await response.json();
                      return { error: errorData.error };
                  }
              }

              $(document).on('paste', function(event) {
                  const clipboardData = event.originalEvent.clipboardData;
                  if (clipboardData && clipboardData.items) {
                      for (let i = 0; i < clipboardData.items.length; i++) {
                          const item = clipboardData.items[i];
                          if (item.kind === 'file') {
                              const pasteFile = item.getAsFile();
                              uploadFile(pasteFile);
                              break;
                          }
                      }
                  }
              });

              async function compressImage(file, quality = 0.5, maxResolution = 20000000) {
                  return new Promise((resolve) => {
                      const image = new Image();
                      image.onload = () => {
                          const width = image.width;
                          const height = image.height;
                          const resolution = width * height;
                          let scale = 1;
                          if (resolution > maxResolution) {
                              scale = Math.sqrt(maxResolution / resolution);
                          }
                          const targetWidth = Math.round(width * scale);
                          const targetHeight = Math.round(height * scale);
                          const canvas = document.createElement('canvas');
                          const ctx = canvas.getContext('2d');
                          canvas.width = targetWidth;
                          canvas.height = targetHeight;
                          ctx.drawImage(image, 0, 0, targetWidth, targetHeight);
                          canvas.toBlob((blob) => {
                              const compressedFile = new File([blob], file.name, { type: 'image/jpeg' });
                              toastr.success('图片压缩成功！');
                              resolve(compressedFile);
                          }, 'image/jpeg', quality);
                      };
                      const reader = new FileReader();
                      reader.onload = (event) => {
                          image.src = event.target.result;
                      };
                      reader.readAsDataURL(file);
                  });
              }

              $('#urlBtn, #bbcodeBtn, #markdownBtn').on('click', function() {
                  const fileLinks = originalImageURLs.map(url => url.trim()).filter(url => url !== '');
                  if (fileLinks.length > 0) {
                      let formattedLinks = '';
                      switch ($(this).attr('id')) {
                          case 'urlBtn':
                              formattedLinks = fileLinks.join('\\n\\n');
                              break;
                          case 'bbcodeBtn':
                              formattedLinks = fileLinks.map(url => '[img]' + url + '[/img]').join('\\n\\n');
                              break;
                          case 'markdownBtn':
                              formattedLinks = fileLinks.map(url => '![image](' + url + ')').join('\\n\\n');
                              break;
                          default:
                              formattedLinks = fileLinks.join('\\n');
                      }
                      $('#fileLink').val(formattedLinks);
                      adjustTextareaHeight($('#fileLink')[0]);
                      copyToClipboardWithToastr(formattedLinks);
                  }
              });

              function handleFileClear(event) {
                  $('#fileLink').val('');
                  adjustTextareaHeight($('#fileLink')[0]);
                  hideButtonsAndTextarea();
                  originalImageURLs = [];
              }

              function adjustTextareaHeight(textarea) {
                textarea.style.height = '1px';
                textarea.style.height = (textarea.scrollHeight > 200 ? 200 : textarea.scrollHeight) + 'px';
                
                if (textarea.scrollHeight > 200) {
                    textarea.style.overflowY = 'auto';
                } else {
                    textarea.style.overflowY = 'hidden';
                }
            }            

              function copyToClipboardWithToastr(text) {
                  const input = document.createElement('textarea');
                  input.value = text;
                  document.body.appendChild(input);
                  input.select();
                  document.execCommand('copy');
                  document.body.removeChild(input);
                  toastr.success('已复制到剪贴板', '', { timeOut: 300 });
              }

              function hideButtonsAndTextarea() {
                  $('#urlBtn, #bbcodeBtn, #markdownBtn, #fileLink').parent('.form-group').hide();
              }

              function saveToLocalCache(url, fileName) {
                  const timestamp = new Date().toLocaleString('zh-CN', { hour12: false });
                  const cacheData = JSON.parse(localStorage.getItem('uploadCache')) || [];
                  cacheData.push({ url, fileName, timestamp });
                  localStorage.setItem('uploadCache', JSON.stringify(cacheData));
              }

              $('#viewCacheBtn').on('click', function() {
                  const cacheData = JSON.parse(localStorage.getItem('uploadCache')) || [];
                  const cacheContent = $('#cacheContent');
                  cacheContent.empty();
                  if (isCacheVisible) {
                      cacheContent.hide();
                      $('#fileLink').val('');
                      $('#fileLink').parent('.form-group').hide();
                      isCacheVisible = false;
                  } else {
                      if (cacheData.length > 0) {
                          cacheData.reverse();
                          cacheData.forEach((item) => {
                              const listItem = $('<div class="cache-item"></div>')
                                  .text(item.timestamp + ' - ' + item.fileName)
                                  .data('url', item.url);
                              cacheContent.append(listItem);
                              cacheContent.append('<br>');
                          });
                          cacheContent.show();
                      } else {
                          cacheContent.append('<div>还没有记录哦！</div>').show();
                      }
                      isCacheVisible = true;
                  }
              });

              $(document).on('click', '.cache-item', function() {
                  const url = $(this).data('url');
                  originalImageURLs = [];
                  $('#fileLink').val('');
                  originalImageURLs.push(url);
                  $('#fileLink').val(originalImageURLs.map(url => url.trim()).join('\\n\\n'));
                  $('.form-group').show();
                  adjustTextareaHeight($('#fileLink')[0]);
              });
          });
      </script>
</body>
</html>  
`, { headers: { 'Content-Type': 'text/html;charset=UTF-8' } });
  await cache.put(cacheKey, response.clone());
  return response;
}

async function handleAdminRequest(DATABASE, request, USERNAME, PASSWORD) {
  if (!authenticate(request, USERNAME, PASSWORD)) {
    return new Response('Unauthorized', { status: 401, headers: { 'WWW-Authenticate': 'Basic realm="Admin"' } });
  }
  return await generateAdminPage(DATABASE);
}

function isValidCredentials(authHeader, USERNAME, PASSWORD) {
  const base64Credentials = authHeader.split(' ')[1];
  const credentials = atob(base64Credentials).split(':');
  const username = credentials[0];
  const password = credentials[1];
  return username === USERNAME && password === PASSWORD;
}

async function generateAdminPage(DATABASE) {
  const mediaData = await fetchMediaData(DATABASE);
  const mediaHtml = mediaData.map(({ url }) => {
    const fileExtension = url.split('.').pop().toLowerCase();
    const timestamp = url.split('/').pop().split('.')[0];
    if (fileExtension === 'mp4') {
      return `
      <div class="media-container" data-key="${url}" onclick="toggleImageSelection(this)">
        <div class="media-type">视频</div>
        <video class="gallery-video" style="width: 100%; height: 100%; object-fit: contain;" data-src="${url}" controls>
          <source src="${url}" type="video/mp4">
          您的浏览器不支持视频标签。
        </video>
        <div class="upload-time">上传时间: ${new Date(parseInt(timestamp)).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}</div>
      </div>
      `;
    } else {
      return `
      <div class="image-container" data-key="${url}" onclick="toggleImageSelection(this)">
        <img data-src="${url}" alt="Image" class="gallery-image lazy">
        <div class="upload-time">上传时间: ${new Date(parseInt(timestamp)).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}</div>
      </div>
      `;
    }
  }).join('');
  
  const html = `
  <!DOCTYPE html>
  <html>
    <head>
      <title>图库</title>
      <link rel="icon" href="https://p1.meituan.net/csc/c195ee91001e783f39f41ffffbbcbd484286.ico" type="image/x-icon">
      <meta name="viewport" content="width=device-width, initial-scale=1.0">
      <style>
      body {
        font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
        background-color: #f4f4f4;
        margin: 0;
        padding: 20px;
      }
      .header {
        position: sticky;
        top: 0;
        background-color: #ffffff;
        z-index: 1000;
        display: flex;
        justify-content: space-between;
        align-items: center;
        margin-bottom: 20px;
        padding: 15px 20px;
        box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1);
        border-radius: 8px;
      }
      .gallery {
        display: grid;
        grid-template-columns: repeat(auto-fill, minmax(200px, 1fr));
        gap: 16px;
      }
      .image-container, .media-container {
        position: relative;
        overflow: hidden;
        border-radius: 12px;
        box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);
        aspect-ratio: 1 / 1;
        transition: transform 0.3s, box-shadow 0.3s;
      }
      .media-type {
        position: absolute;
        top: 10px;
        left: 10px;
        background-color: rgba(0, 0, 0, 0.7);
        color: white;
        padding: 5px;
        border-radius: 5px;
        font-size: 14px;
        z-index: 10;
        cursor: pointer;
      }
      .image-container .upload-time, .media-container .upload-time {
        position: absolute;
        bottom: 10px;
        left: 10px;
        background-color: rgba(255, 255, 255, 0.7);
        padding: 5px;
        border-radius: 5px;
        color: #000;
        font-size: 14px;
        z-index: 10;
        display: none;
      }
      .image-container:hover, .media-container:hover {
        transform: scale(1.05);
        box-shadow: 0 4px 16px rgba(0, 0, 0, 0.2);
      }
      .gallery-image {
        width: 100%;
        height: 100%;
        object-fit: contain;
        transition: opacity 0.3s;
        opacity: 0;
      }
      .gallery-image.loaded {
        opacity: 1;
      }
      .media-container.selected, .image-container.selected {
        border: 2px solid #007bff;
        background-color: rgba(0, 123, 255, 0.1);
      }
      .footer {
        margin-top: 20px;
        text-align: center;
        font-size: 18px;
        color: #555;
      }
      .delete-button {
        background-color: #ff4d4d;
        color: white;
        border: none;
        border-radius: 5px;
        padding: 10px 15px;
        cursor: pointer;
        transition: background-color 0.3s;
        width: auto;
      }
      .delete-button:hover {
        background-color: #ff1a1a;
      }
      .hidden {
        display: none;
      }
      @media (max-width: 600px) {
        .gallery {
          grid-template-columns: repeat(2, 1fr);
        }
        .header {
          flex-direction: row;
          align-items: center;
        }
        .header-right {
          margin-left: auto;
        }
        .footer {
          font-size: 16px;
        }
        .delete-button {
          width: 100%;
          margin-top: 10px;
        }
      }
      </style>
      <script>
        let selectedCount = 0;
        const selectedKeys = new Set();
        function toggleImageSelection(container) {
          const key = container.getAttribute('data-key');
          container.classList.toggle('selected');
          const uploadTime = container.querySelector('.upload-time');
          if (container.classList.contains('selected')) {
            selectedKeys.add(key);
            selectedCount++;
            uploadTime.style.display = 'block';
          } else {
            selectedKeys.delete(key);
            selectedCount--;
            uploadTime.style.display = 'none';
          }
          updateDeleteButton();
        }
        function updateDeleteButton() {
          const deleteButton = document.getElementById('delete-button');
          const countDisplay = document.getElementById('selected-count');
          countDisplay.textContent = selectedCount;
          const headerRight = document.querySelector('.header-right');
          if (selectedCount > 0) {
            headerRight.classList.remove('hidden');
          } else {
            headerRight.classList.add('hidden');
          }
        }
        async function deleteSelectedImages() {
          if (selectedKeys.size === 0) return;
          const response = await fetch('/delete-images', {
            method: 'POST',
            headers: {
              'Content-Type': 'application/json'
            },
            body: JSON.stringify(Array.from(selectedKeys))
          });
          if (response.ok) {
            alert('选中的媒体已删除');
            location.reload();
          } else {
            alert('删除失败');
          }
        }
        document.addEventListener('DOMContentLoaded', () => {
          const images = document.querySelectorAll('.gallery-image[data-src]');
          const options = {
            root: null,
            rootMargin: '0px',
            threshold: 0.1
          };
          const imageObserver = new IntersectionObserver((entries, observer) => {
            entries.forEach(entry => {
              if (entry.isIntersecting) {
                const img = entry.target;
                img.src = img.dataset.src;
                img.onload = () => img.classList.add('loaded');
                observer.unobserve(img);
              }
            });
          }, options);
          images.forEach(image => {
            imageObserver.observe(image);
          });
        });
      </script>
    </head>
    <body>
      <div class="header">
        <div class="header-left">
          <span>媒体文件 ${mediaData.length} 个</span>
          <span>已选中: <span id="selected-count">0</span>个</span>
        </div>
        <div class="header-right hidden">          
          <button id="delete-button" class="delete-button" onclick="deleteSelectedImages()">删除选中</button>
        </div>
      </div>
      <div class="gallery">
        ${mediaHtml}
      </div>
      <div class="footer">
        到底啦
      </div>
    </body>
  </html>
  `;
  return new Response(html, { status: 200, headers: { 'Content-Type': 'text/html; charset=utf-8' } });
}

async function fetchMediaData(DATABASE) {
  const result = await DATABASE.prepare('SELECT url, fileId FROM media').all();
  const mediaData = result.results.map(row => {
    const timestamp = parseInt(row.url.split('/').pop().split('.')[0]);
    return { fileId: row.fileId, url: row.url, timestamp: timestamp };
  });
  mediaData.sort((a, b) => b.timestamp - a.timestamp);
  return mediaData.map(({ fileId, url }) => ({ fileId, url }));
}

async function handleUploadRequest(request, DATABASE, enableAuth, USERNAME, PASSWORD, domain, TG_BOT_TOKEN, TG_CHAT_ID) {
  try {
    const formData = await request.formData();
    const file = formData.get('file');
    if (!file) throw new Error('缺少文件');
    if (enableAuth && !authenticate(request, USERNAME, PASSWORD)) {
      return new Response('Unauthorized', { status: 401, headers: { 'WWW-Authenticate': 'Basic realm="Admin"' } });
    }
    const uploadFormData = new FormData();
    uploadFormData.append("chat_id", TG_CHAT_ID);
    let fileId;
    if (file.type.startsWith('image/gif')) {
      const newFileName = file.name.replace(/\.gif$/, '.jpeg');
      const newFile = new File([file], newFileName, { type: 'image/jpeg' });
      uploadFormData.append("document", newFile);
    } else {
      uploadFormData.append("document", file);
    }
    const telegramResponse = await fetch(`https://api.telegram.org/bot${TG_BOT_TOKEN}/sendDocument`, { method: 'POST', body: uploadFormData });
    if (!telegramResponse.ok) {
      const errorData = await telegramResponse.json();
      throw new Error(errorData.description || '上传到 Telegram 失败');
    }
    const responseData = await telegramResponse.json();
    if (responseData.result.video) fileId = responseData.result.video.file_id;
    else if (responseData.result.document) fileId = responseData.result.document.file_id;
    else if (responseData.result.sticker) fileId = responseData.result.sticker.file_id;
    else throw new Error('返回的数据中没有文件 ID');
    const fileExtension = file.name.split('.').pop();
    const timestamp = Date.now();
    const imageURL = `https://${domain}/${timestamp}.${fileExtension}`;
    await DATABASE.prepare('INSERT INTO media (url, fileId) VALUES (?, ?) ON CONFLICT(url) DO NOTHING').bind(imageURL, fileId).run();
    return new Response(JSON.stringify({ data: imageURL }), { status: 200, headers: { 'Content-Type': 'application/json' } });
  } catch (error) {
    console.error('内部服务器错误:', error);
    return new Response(JSON.stringify({ error: error.message }), { status: 500, headers: { 'Content-Type': 'application/json' } });
  }
}

async function handleImageRequest(request, DATABASE, TG_BOT_TOKEN) {
  const requestedUrl = request.url;
  const cache = caches.default;
  const cacheKey = new Request(requestedUrl);
  const cachedResponse = await cache.match(cacheKey);
  if (cachedResponse) return cachedResponse;
  const result = await DATABASE.prepare('SELECT fileId FROM media WHERE url = ?').bind(requestedUrl).first();
  if (result) {
    const fileId = result.fileId;
    const getFileResponse = await fetch(`https://api.telegram.org/bot${TG_BOT_TOKEN}/getFile?file_id=${fileId}`);
    if (!getFileResponse.ok) return new Response(null, { status: 404 });
    const fileData = await getFileResponse.json();
    const filePath = fileData.result.file_path;
    const telegramFileUrl = `https://api.telegram.org/file/bot${TG_BOT_TOKEN}/${filePath}`;
    const response = await fetch(telegramFileUrl);
    if (response.ok) {
      const fileExtension = requestedUrl.split('.').pop().toLowerCase();
      let contentType = 'text/plain';
      if (fileExtension === 'jpg' || fileExtension === 'jpeg') contentType = 'image/jpeg';
      if (fileExtension === 'png') contentType = 'image/png';
      if (fileExtension === 'gif') contentType = 'image/gif';
      if (fileExtension === 'webp') contentType = 'image/webp';
      if (fileExtension === 'mp4') contentType = 'video/mp4';
      const headers = new Headers(response.headers);
      headers.set('Content-Type', contentType);
      headers.set('Content-Disposition', 'inline');
      const responseToCache = new Response(response.body, { status: response.status, headers });
      await cache.put(cacheKey, responseToCache.clone());
      return responseToCache;
    }
  }
  const notFoundResponse = new Response(null, { status: 404 });
  await cache.put(cacheKey, notFoundResponse.clone());
  return notFoundResponse;
}

async function handleBingImagesRequest(request) {
  const cache = caches.default;
  const cacheKey = new Request('https://cn.bing.com/HPImageArchive.aspx?format=js&idx=0&n=5');
  const cachedResponse = await cache.match(cacheKey);
  if (cachedResponse) return cachedResponse;
  const res = await fetch(cacheKey);
  const bingData = await res.json();
  const images = bingData.images.map(image => ({ url: `https://www.dmoe.cc/random.php` }));
  const returnData = { status: true, message: "操作成功", data: images };
  const response = new Response(JSON.stringify(returnData), { status: 200, headers: { 'Content-Type': 'application/json' } });
  await cache.put(cacheKey, response.clone());
  return response;
}

async function handleDeleteImagesRequest(request, DATABASE) {
  if (request.method !== 'POST') {
    return new Response('Method Not Allowed', { status: 405 });
  }
  try {
    const keysToDelete = await request.json();
    if (!Array.isArray(keysToDelete) || keysToDelete.length === 0) {
      return new Response(JSON.stringify({ message: '没有要删除的项' }), { status: 400 });
    }
    const placeholders = keysToDelete.map(() => '?').join(',');
    const result = await DATABASE.prepare(`DELETE FROM media WHERE url IN (${placeholders})`).bind(...keysToDelete).run();
    if (result.changes === 0) {
      return new Response(JSON.stringify({ message: '未找到要删除的项' }), { status: 404 });
    }
    const cache = caches.default;
    for (const url of keysToDelete) {
      const cacheKey = new Request(url);
      const cachedResponse = await cache.match(cacheKey);
      if (cachedResponse) {
        await cache.delete(cacheKey);
      }
    }
    return new Response(JSON.stringify({ message: '删除成功' }), { status: 200 });
  } catch (error) {
    return new Response(JSON.stringify({ error: '删除失败', details: error.message }), { status: 500 });
  }
}

三、关于背景修改

下面是两个可用的动漫背景随机图，可以提供修改。

1	const images = bingData.images.map(image => ({ url: `https://www.dmoe.cc/random.php` }));

这行代码里面的url可以改掉就行了。

动漫
www.dmoe.cc/random.php
api.mtyqx.cn/tapi/random.php （挺好看的动漫）

分享几个公用的高清壁纸图片的API | LearnKu 产品论坛

四、关于PicGo

到插件中下载一个【自定义图床】插件，按照下面设置即可

Netcup注册

Posted on 2024-10-25

Netcup可以注册后缀为de的域名一年1.32欧元，续费一样相当划算

另外Netcup有0.84欧元/月的VPS套餐，每个账户只允许申请一台。

探针如下：

1.家宽直连，最好在下午4点到早上6点去试，这样能快一点，德国人在上班。用这个链接去买，感谢justwe佬的分享，应该不是aff

1
2
3

1o：https://www.netcup.de/bestellen/produkt.php?produkt=3422
2o：https://www.netcup.de/bestellen/produkt.php?produkt=3423
4o：https://www.netcup.de/bestellen/produkt.php?produkt=3424

2.一直往下点就行，不行就翻译一下，选择注册帐号，填入你的真实信息，地址不能有“，”全英文，地址选中国，公司可以不填
3.会发一封邮件说工作人员正在审核，问你要不要付款，你可以取消之类的，如果你用家宽的话等5分钟左右会收到两封连在一起的邮件，一封给了你帐号密码，另一封让你去验证身份，你打开验证身份即可，注意一定要在第一次就选择预付款，不要选验证身份
4.去https://www.servercontrolpanel.de/SCP/Home 登录，会发二步验证到你的邮箱，去billing付款，信用卡或者paypal，等机器开通，done！

全程顺利的话半小时解决，我等机器开通等了1个小时，现在只要付钱就能过了，方便多了

findbook的使用方法

Posted on 2024-09-19

我们在searching图书的时候，下载时如果长时间停留0%，那一定是你的网关设置的有问题，只需要按照下面配置即可。

访问下面地址，查看自己的网络状态

https://ipfs.github.io/public-gateway-checker/

将域名复制编辑

https://ipfs.runfission.com
https://4everland.io
https://gateway.pinata.cloud
https://trustless-gateway.link

将上面的内容复制到页面右上角的⚙️中去，然后保存。

重新尝试下载

HSD高频航班累订阅事件说明-ICKI

Posted on 2024-05-21

8类基础数据

Mockjs 造数据写入Mysql

Posted on 2024-05-20

Mock.js 是一个用于生成随机数据和拦截 Ajax 请求的库，常用于前端开发中对接口进行模拟。通过 Mock.js，可以快速生成各种类型的随机数据，包括但不限于文本、数字、日期、布尔值等，从而方便地进行数据模拟和测试。除了数据生成外，Mock.js 也可以拦截前端发送的 Ajax 请求，返回预先设定的模拟数据，使前端开发人员能够在没有后端支持的情况下进行接口调试及开发。

下面我们使用Mockjs生成数据，写入到Mysql数据库

下面是项目目录结构：

.
├── main.js
├── package.json
└── util
    ├── dbutil.js
    └── mockutil.js

提前使用npm或cnpm安装所需要的依赖:

1 2	cnpm install mockjs --save cnpm install mysql --save

数据库建表


CREATE TABLE aoc_flight_info_dw (
    id                  BIGINT AUTO_INCREMENT COMMENT '主键' PRIMARY KEY,
    flight_date         DATE                                 NOT NULL COMMENT '航班日期',
    flight_no           VARCHAR(10)                          NOT NULL COMMENT '航班号',
    dep_airport         VARCHAR(4)                           NOT NULL COMMENT '起飞机场',
    arr_airport         VARCHAR(4)                           NOT NULL COMMENT '到达机场',
    transfer_number     INT(10)    DEFAULT 0                 NOT NULL COMMENT '中转人数',
    etd                 DATETIME                             NOT NULL COMMENT '预计起飞时间',
    eta                 DATETIME                             NOT NULL COMMENT '预计到达时间',
    exist_vip           TINYINT(1) DEFAULT 0                 NOT NULL COMMENT '是否有VIP旅客',
    previous_delay      TINYINT(1) DEFAULT 0                 NOT NULL COMMENT '前段航班是否延误',
    estimate_delay_time INT(10)    DEFAULT 0                 NOT NULL COMMENT '预计延误时间',
    std                 DATETIME                             NOT NULL COMMENT '计划起飞时间',
    sta                 DATETIME                             NOT NULL COMMENT '计划到达时间',
    flight_type         VARCHAR(10)                          NOT NULL COMMENT '航班性质',
    d_or_i              VARCHAR(10)                          NOT NULL COMMENT '国际/国内/地区',
    ac_type             VARCHAR(10)                          NOT NULL COMMENT '机型',
    ac_reg              VARCHAR(10)                          NOT NULL COMMENT '机号',
    warn_level          VARCHAR(10)                          NULL COMMENT '预警航班等级',
    create_time         DATETIME   DEFAULT current_timestamp NOT NULL COMMENT '创建时间',
    update_time         DATETIME   DEFAULT current_timestamp NOT NULL ON UPDATE current_timestamp COMMENT '更新时间',
    CONSTRAINT unq_flight_idx UNIQUE (flight_date, flight_no, dep_airport, arr_airport)
) COMMENT '航班信息表同步数仓' CHARSET = utf8mb4;

CREATE INDEX aoc_flight_info_flight_date_idx ON aoc_flight_info_dw (flight_date, ac_reg);


CREATE TABLE aoc_passenger_info_dw (
    id             BIGINT AUTO_INCREMENT COMMENT '主键' PRIMARY KEY,
    aoc_flight_id  BIGINT                                 NOT NULL COMMENT '航班信息表id',
    passenger_name VARCHAR(100) DEFAULT ''                NOT NULL COMMENT '姓名',
    vip_type       VARCHAR(10)  DEFAULT ''                NOT NULL COMMENT 'VIP类型',
    vip_line       VARCHAR(100) DEFAULT ''                NOT NULL COMMENT 'VIP航程',
    create_time    DATETIME     DEFAULT current_timestamp NOT NULL COMMENT '创建时间',
    update_time    DATETIME     DEFAULT current_timestamp NOT NULL ON UPDATE current_timestamp COMMENT '更新时间'
) COMMENT 'VIP旅客信息表数仓同步' CHARSET = utf8mb4;

CREATE INDEX aoc_passenger_info_aoc_flight_id_idx ON aoc_passenger_info_dw (aoc_flight_id);

1. 连接Mysql数据库

var Mysql = require('mysql');

var conn = Mysql.createConnection({
    host   : '172.22.17.38',
    user   : 'u_tqd',
    password : '62w5LjLi',
    port: '3306',
    database: 'tqddb',
});

exports.insert = function(sql, dim_arr){
    conn.connect();
    dim_arr.data.forEach(element=>{
        conn.query(sql, element, function(err, result){
            if(err){
                console.log('err:', err.message);
            }
            console.log('------INSERT-------');
            console.log('res:', result);
            console.log('------SUCCESS-------');
        });
    })
    conn.end();
}

2. 使用mock伪造数据

var Mock = require('mockjs');

exports.aoc_flight_info_dw = Mock.mock({
    "data|500": [
      [
            "@increment(1)",                    // id
            "@date('yyyy-MM-dd')",              // flight_date
            /^HO[0-9]{4}$/,                     // flight_no
            /^[A-Z]{3}$/,                       // dep_airport
            /^[A-Z]{3}$/,                       // arr_airport
            "@integer(0, 20 )",                 // transfer_number
            "@datetime('yyyy-MM-dd HH:mm:ss')", // etd
            "@datetime('yyyy-MM-dd HH:mm:ss')", // eta
            "@integer(0, 1 )",                  // exist_vip
            "@integer(0, 1 )",                  // previous_delay
            "@integer(1800, 18000 )",           // estimate_delay_time
            "@datetime('yyyy-MM-dd HH:mm:ss')", // std
            "@datetime('yyyy-MM-dd HH:mm:ss')", // sta
            "@integer(0, 3 )",                  // flight_type
            /D|I|R/,                            // d_or_i
            /A320|A321|B787|A319/,              // ac_type
            /^[0-9]{4}$/,                       // ac_reg
            "@integer(0, 1 )",                  // warn_level
            "@datetime('yyyy-MM-dd HH:mm:ss')", // create_time
            "@datetime('yyyy-MM-dd HH:mm:ss')", // update_time
        ]
    ]
});

exports.aoc_passenger_info_dw = Mock.mock({
    "data|500": [
      [
         "@increment(1)",                   // id
         "@integer(0, 1000 )",              // aoc_flight_id
         "@cname()",                        // passenger_name
         "@integer(0, 3 )",                 // vip_type
         "@integer(0, 3 )",                 // vip_line
         "@datetime('yyyy-MM-dd HH:mm:ss')",// create_time
         "@datetime('yyyy-MM-dd HH:mm:ss')",// update_time
        ]
    ]
});

3. 编写main

var mockutil = require('./util/mockutil.js');
var dbutil = require('./util/dbutil.js');

function aoc_flight_info_dw(){
    var sql = 'INSERT INTO tqddb.aoc_flight_info_dw (id, flight_date, flight_no, dep_airport, arr_airport, transfer_number, etd, eta, exist_vip, previous_delay, estimate_delay_time, std, sta, flight_type, d_or_i, ac_type, ac_reg, warn_level, create_time, update_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);'
    dbutil.insert(sql, mockutil.aoc_flight_info_dw);
}

function aoc_passenger_info_dw(){
    var sql = 'INSERT INTO tqddb.aoc_passenger_info_dw (id, aoc_flight_id, passenger_name, vip_type, vip_line, create_time, update_time) VALUES (?,?,?,?,?,?,?);'
    dbutil.insert(sql, mockutil.aoc_passenger_info_dw);
}

function main(){
    //aoc_flight_info_dw();
    aoc_passenger_info_dw()

}

main();

K2K Template 脚本

Posted on 2024-05-17

使用flink 同步Kafka到Kafka

-- 前两段是kafka2kafka
-- LDP init的话用目录下面的python脚本
CREATE TABLE kafka_source_ldp (
    a STRING
)with ( 'connector' = 'kafka',
      'topic' = 'LOADSHEET.LDP',
      'scan.startup.mode' = 'earliest-offset',
      'properties.group.id' = 'pipeline_ums',
      'properties.bootstrap.servers' = 'jxbigdatakafka01.juneyaoair.com:9092,jxbigdatakafka02.juneyaoair.com:9092,jxbigdatakafka03.juneyaoair.com:9092',
      'properties.fetch.max.bytes' = '5242880',
      'properties.allow.auto.create.topics' = 'false',
      'properties.enable.auto.commit' = 'true',
      'format' = 'raw' );

CREATE TABLE kafka_sink_ldp (
    a STRING
)with ( 'connector' = 'kafka',
      'topic' = 'JXHK.AUTO.DC_AOC_LOADSHEET_LDP',
      'properties.bootstrap.servers' = '172.27.2.3:9092,172.27.2.4:9092,172.27.2.10:9092',
      'properties.allow.auto.create.topics' = 'false',
      'properties.enable.auto.commit' = 'true',
      'format' = 'raw' );

insert into kafka_sink_ldp
select * from kafka_source_ldp;

Doris关联Hive表

Posted on 2024-05-15

Doris关联Hive外部数据源

元数据创建

CREATE CATALOG test_hive PROPERTIES (
    'type'='hms',
    'hive.metastore.uris' = 'thrift://172.22.17.20:9083,thrift://172.22.17.21:9083,thrift://172.22.17.22:9083',
    'hadoop.username' = 'hive',
    'dfs.nameservices' = 'Tdsop',
    'dfs.ha.namenodes.Tdsop' = 'nn1,nn2',
    'dfs.namenode.rpc-address.Tdsop.nn1' = '172.22.17.20:8022',
    'dfs.namenode.rpc-address.Tdsop.nn2' = '172.22.17.21:8022',
    'dfs.client.failover.proxy.provider.Tdsop' = 'org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider'
);

常用命令

SHOW catalogs;
SHOW CATALOG test_hive;
SHOW DATABASES FROM test_hive;
SHOW TABLES FROM test_hive.ods;
SELECT * FROM test_hive.ods.ce_flight;

HSD梳理

Posted on 2024-05-15

HSD名称及英文简称映射

名称	英文简称	参考黑屏指令
航班计划数据	iSCH	ST
库存数据	iINV	RO
客票数据	iETS	DETR
EMD票数据	iEMD	DEMD
订座PNR数据	iPNR	RT
订座PNR数据(历史)	iPNRH	RTC
离港旅客数据	iCKI	PR
离港航班数据	iFLT	SY

中转航班的逻辑

-- 按 照票号+出票日期 取最新的数据
create table test.djj_temp_hsd_iets as
select 
    *
from  
(
select 
    a.stamp,
    a.dttm,
    a.ticketidentification_tktissuedate,
    a.ticketimage_ticketnumber,
    a.itinerary_ticketcoupon,
    row_number() over(partition by a.ticketidentification_tktissuedate, a.ticketimage_ticketnumber order by a.dttm desc) r
from 
    ods.cst_hsd_iets a
) t  
where  r=1 
;
    
    
-- 拆分json
drop table test.djj_temp_hsd_iets_01;    
create table test.djj_temp_hsd_iets_01 as
select 
    a.stamp,
    a.dttm,
    a.ticketidentification_tktissuedate issuedate, -- 出票日期
    substr(a.ticketimage_ticketnumber,1,13) ticketnumber, --票号
    get_json_object(itinerary, '$.CouponNumber') couponNumber,
    get_json_object(itinerary, '$.CouponStatus') couponStatus, -- 客票状态
    get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Carrier'), '$.AirlineCode') carrier_airlinecode, 
    get_json_object(get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Carrier'), '$.FlightNumber'), '$.Value') carrier_flightnumber,
    get_json_object(get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Carrier'), '$.FlightNumber'), '$.Suffix') carrier_flightnumber_suffix,
    get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.OperatingCarrier'), '$.AirlineCode') operating_airlinecode, -- 承运航司
    get_json_object(get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.OperatingCarrier'), '$.FlightNumber'), '$.Value') operating_flightnumber, -- 承运航班号
    get_json_object(get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.OperatingCarrier'), '$.FlightNumber'), '$.Suffix') operating_flightnumber_suffix, -- 承运航班号后缀
    get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Departure'), '$.Date') departure_date,
    get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Departure'), '$.AirportCode') departure_airportcode,
    get_json_object(get_json_object(get_json_object(itinerary, '$.Flight'), '$.Arrival'), '$.AirportCode') arrival_airportcode,
    get_json_object(get_json_object(get_json_object(itinerary, '$.Open'), '$.OpenDeparture'), '$.AirportCode') open_departure_airportcode,
    get_json_object(get_json_object(get_json_object(itinerary, '$.Open'), '$.OpenArrival'), '$.AirportCode') open_arrival_airportcode,
    get_json_object(get_json_object(itinerary, '$.Open'), '$.OperatingCarrier') open_operatingcarrier
from 
    test.djj_temp_hsd_iets a
    LATERAL VIEW explode(split(
     regexp_replace(
         regexp_replace(
            itinerary_ticketcoupon, 
            '\\[|\\]' , ''),-- 将json数组两边的中括号去掉
            
              '\\}\\,\\{' , '\\}\\;\\{'),-- 将json数组元素之间的逗号换成分号
                
                 '\\;')-- 以分号作为分隔符(split函数以分号作为分隔)
          )  a as itinerary
    ;
    

-- 合并出发到达
drop table test.djj_temp_hsd_iets_02;    
create table test.djj_temp_hsd_iets_02 as    
select 
    a.issuedate,
    a.ticketnumber,
    a.couponnumber,
    a.couponstatus,
    coalesce(a.departure_airportcode, a.open_departure_airportcode) departure_airportcode,
    coalesce(a.arrival_airportcode, a.open_arrival_airportcode) arrival_airportcode
from  
    test.djj_temp_hsd_iets_01 a
    ;
    

-- 匹配出发到达城市    
drop table test.djj_temp_hsd_iets_03;    
create table test.djj_temp_hsd_iets_03 as     
select 
    a.issuedate,
    a.ticketnumber,
    a.couponnumber,
    a.couponstatus,
    a.departure_airportcode,
    a.arrival_airportcode,
    b.city_code departure_citycode,
    c.city_code arrival_citycode,
    coalesce(b.city_code, a.departure_airportcode) departure, -- 以防有的机场匹配不到城市
    coalesce(c.city_code, a.arrival_airportcode) arrival -- 以防有的机场匹配不到城市
from 
    test.djj_temp_hsd_iets_02  a 
left join 
    ods.t_airport_info_zp b on a.departure_airportcode=b.airport_code and b.dp='ACTIVE'
left join 
    ods.t_airport_info_zp c on a.arrival_airportcode=c.airport_code and c.dp='ACTIVE'
where 
    a.couponstatus<>'V'  -- 排除废票
    ;
    


-- 匹配出发到达城市    
drop table test.djj_temp_hsd_iets_04;    
create table test.djj_temp_hsd_iets_04 as      
select 
    t.issuedate,
    t.ticketnumber,
    t.seg_num,
    substr(origin,1,3) origin,
    substr(destination,-3,3) destination
from  
(
select 
    a.issuedate,
    a.ticketnumber,
    sum(1) over(partition by  a.issuedate, a.ticketnumber) seg_num, -- 总航段数
    concat_ws(',', collect_list(departure) over(partition by a.issuedate, a.ticketnumber order by a.couponnumber)) as origin,
    concat_ws(',', collect_list(arrival) over(partition by a.issuedate, a.ticketnumber order by a.couponnumber)) as destination,
    row_number() over(partition by a.issuedate, a.ticketnumber order by a.couponnumber desc) as r
from  
    test.djj_temp_hsd_iets_03 a
) t 
where 
    r=1
    and substr(origin,1,3)<>substr(destination,-3,3) -- 排除往返票
    and seg_num>1 -- 筛选多段的
    ;
    
    
select 
    issuedate,
    ticketnumber,
    seg_num,
    origin,
    destination
from  
    test.djj_temp_hsd_iets_04 a
    ;

FlinkSQL Kafka2Doris Table program cannot be compiled This is a bug. Please file an issue

Posted on 2024-05-07

错误日志:

2024-05-07 17:34:33,817 INFO  org.apache.doris.flink.table.DorisDynamicTableSink           [] - Send request to Doris FE 'http://10.17.23.55:8030/api/ods/ods_fin_nc_bd_psndoc/_schema' with user 'juneyaoair_etl'.
2024-05-07 17:34:34,027 INFO  org.apache.doris.flink.sink.DorisSink                        [] - Send request to Doris FE 'http://10.17.23.55:8030/api/ods/ods_fin_nc_bd_psndoc/_schema' with user 'juneyaoair_etl'.
2024-05-07 17:35:09,997 WARN  org.apache.flink.table.runtime.generated.GeneratedClass      [] - Failed to compile split code, falling back to original code
org.apache.flink.util.FlinkRuntimeException: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:94) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:97) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.getClass(GeneratedClass.java:120) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.operators.CodeGenOperatorFactory.getStreamOperatorClass(CodeGenOperatorFactory.java:51) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.preValidate(StreamingJobGraphGenerator.java:498) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:221) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:153) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:1024) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:56) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:43) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:59) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:72) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2188) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:206) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95) ~[flink-table-planner_2.12-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:850) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:827) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:108) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at com.juneyaoair.flink.streaming.core.JobApplication.main(JobApplication.java:82) ~[flink-streaming-core-1.16.2.jar:2.1.6-SNAPSHOT]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_251]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_251]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_251]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_251]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:843) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1087) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1165) ~[flink-dist-1.16.2.jar:1.16.2]
	at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_251]
	at javax.security.auth.Subject.doAs(Subject.java:422) [?:1.8.0_251]
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875) [hadoop-common.jar:?]
	at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) [flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1165) [flink-dist-1.16.2.jar:1.16.2]
Caused by: org.apache.flink.shaded.guava30.com.google.common.util.concurrent.UncheckedExecutionException: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2051) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	... 34 more
Caused by: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:107) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4864) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	... 34 more
Caused by: org.codehaus.commons.compiler.CompileException: Line 2, Column 19: Constant pool for class StreamExecCalc$131226 has grown past JVM limit of 0xFFFF
	at org.codehaus.janino.UnitCompiler.compileUnit(UnitCompiler.java:380) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:237) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.compileToClassLoader(SimpleCompiler.java:465) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:216) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:207) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:80) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:75) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:104) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4864) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	... 34 more
Caused by: org.codehaus.janino.util.ClassFile$ClassFileException: Constant pool for class StreamExecCalc$131226 has grown past JVM limit of 0xFFFF
	at org.codehaus.janino.util.ClassFile.addToConstantPool(ClassFile.java:537) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.util.ClassFile.addConstantUtf8Info(ClassFile.java:481) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.util.ClassFile.addMethodInfo(ClassFile.java:600) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:3222) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1357) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1330) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:822) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:432) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$400(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$2.visitPackageMemberClassDeclaration(UnitCompiler.java:411) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$2.visitPackageMemberClassDeclaration(UnitCompiler.java:406) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.Java$PackageMemberClassDeclaration.accept(Java.java:1414) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:406) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.compileUnit(UnitCompiler.java:378) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:237) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.compileToClassLoader(SimpleCompiler.java:465) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:216) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:207) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:80) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:75) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:104) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4864) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	... 34 more
2024-05-07 17:37:08,897 ERROR com.juneyaoair.flink.streaming.core.JobApplication           [] - 任务执行失败：
org.apache.flink.table.api.TableException: Failed to execute sql
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:867) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:827) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:108) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	at com.juneyaoair.flink.streaming.core.JobApplication.main(JobApplication.java:82) ~[flink-streaming-core-1.16.2.jar:2.1.6-SNAPSHOT]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_251]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_251]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_251]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_251]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:843) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1087) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1165) ~[flink-dist-1.16.2.jar:1.16.2]
	at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_251]
	at javax.security.auth.Subject.doAs(Subject.java:422) [?:1.8.0_251]
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875) [hadoop-common.jar:?]
	at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) [flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1165) [flink-dist-1.16.2.jar:1.16.2]
Caused by: org.apache.flink.util.FlinkRuntimeException: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:94) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.getClass(GeneratedClass.java:120) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.operators.CodeGenOperatorFactory.getStreamOperatorClass(CodeGenOperatorFactory.java:51) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.preValidate(StreamingJobGraphGenerator.java:498) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:221) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:153) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:1024) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:56) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:43) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:59) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:72) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2188) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:206) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95) ~[flink-table-planner_2.12-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:850) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	... 19 more
Caused by: org.apache.flink.shaded.guava30.com.google.common.util.concurrent.UncheckedExecutionException: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2051) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.getClass(GeneratedClass.java:120) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.operators.CodeGenOperatorFactory.getStreamOperatorClass(CodeGenOperatorFactory.java:51) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.preValidate(StreamingJobGraphGenerator.java:498) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:221) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:153) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:1024) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:56) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:43) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:59) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:72) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2188) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:206) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95) ~[flink-table-planner_2.12-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:850) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	... 19 more
Caused by: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue.
	at org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:107) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4864) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache.get(LocalCache.java:3962) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.shaded.guava30.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4859) ~[java-udf-warehouse.jar:?]
	at org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.generated.GeneratedClass.getClass(GeneratedClass.java:120) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.table.runtime.operators.CodeGenOperatorFactory.getStreamOperatorClass(CodeGenOperatorFactory.java:51) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.preValidate(StreamingJobGraphGenerator.java:498) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:221) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:153) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:1024) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:56) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:43) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:59) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:72) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2188) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:206) ~[flink-dist-1.16.2.jar:1.16.2]
	at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95) ~[flink-table-planner_2.12-1.16.2.jar:1.16.2]
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:850) ~[flink-table-api-java-uber-1.16.2.jar:1.16.2]
	... 19 more
Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded
	at java.util.HashMap.newNode(HashMap.java:1750) ~[?:1.8.0_251]
	at java.util.HashMap.putVal(HashMap.java:631) ~[?:1.8.0_251]
	at java.util.HashMap.putMapEntries(HashMap.java:515) ~[?:1.8.0_251]
	at java.util.HashMap.putAll(HashMap.java:785) ~[?:1.8.0_251]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3660) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$5800(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitLocalVariableDeclarationStatement(UnitCompiler.java:3545) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitLocalVariableDeclarationStatement(UnitCompiler.java:3513) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.Java$LocalVariableDeclarationStatement.accept(Java.java:3522) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3512) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3569) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$4700(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitBlock(UnitCompiler.java:3531) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitBlock(UnitCompiler.java:3513) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.Java$Block.accept(Java.java:2779) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3512) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3603) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$5100(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitIfStatement(UnitCompiler.java:3535) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitIfStatement(UnitCompiler.java:3513) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.Java$IfStatement.accept(Java.java:2950) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3512) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3569) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$4700(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitBlock(UnitCompiler.java:3531) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitBlock(UnitCompiler.java:3513) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.Java$Block.accept(Java.java:2779) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3512) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.buildLocalVariableMap(UnitCompiler.java:3603) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler.access$5100(UnitCompiler.java:215) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitIfStatement(UnitCompiler.java:3535) ~[flink-table-runtime-1.16.2.jar:1.16.2]
	at org.codehaus.janino.UnitCompiler$12.visitIfStatement(UnitCompiler.java:3513) ~[flink-table-runtime-1.16.2.jar:1.16.2]

原脚本：

-- 第二批
CREATE TABLE KAFKA_SOURCE_FIN_NC_BD_PSNDOC (
    op_type String,
    op_ts String,
    op_ts_time as cast(op_ts as timestamp(3)),
    current_ts String,
    pos String,
    before row<`PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` String,
    `AMCODE` String,
    `TS` String,
    `DR` String,
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` String,
    `TBM_PROP` String,
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String >,
    after row<`PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` String,
    `AMCODE` String,
    `TS` String,
    `DR` String,
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` String,
    `TBM_PROP` String,
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String >,
    data_row as
    case
    when op_type = 'D' then before
    else after
    end,
    watermark for op_ts_time as op_ts_time - interval '5' second
) with (
    'connector' = 'kafka',
    'topic' = 'SEM.NC.JXNC57.BD_PSNDOC',
    'scan.startup.mode' = 'earliest-offset',
    'properties.group.id' = 'pipeline_kafka2doris_3_test',
    'properties.bootstrap.servers' = 'jxbigdatakafka01.juneyaoair.com:9092,jxbigdatakafka02.juneyaoair.com:9092,jxbigdatakafka03.juneyaoair.com:9092',
    'properties.fetch.max.bytes' = '5242880',
    'properties.allow.auto.create.topics' = 'false',
    'properties.enable.auto.commit' = 'true',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username="bigdatauser" password="bigdata@gd66";',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'format' = 'json'
);

CREATE TABLE DORIS_SINK_FIN_NC_BD_PSNDOC (
    `PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` DECIMAL,
    `AMCODE` String,
    `TS` String,
    `DR` DECIMAL(10,0),
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` DECIMAL(38,0),
    `TBM_PROP` DECIMAL(38,0),
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String,
    `OP_TYPE` String,
    `OP_TS` String,
    `OGG_TIME` String,
    `GMT_CREATE` TIMESTAMP,
    PRIMARY KEY (`PK_PSNDOC`) NOT ENFORCED
) with (
    'connector' = 'doris',
    'fenodes' = '10.17.23.55:8030',
    'table.identifier' = 'ods.ods_fin_nc_bd_psndoc',
    'username' = 'juneyaoair_etl',
    'password' = 'juneyaoair_etl123!'
);

insert into DORIS_SINK_FIN_NC_BD_PSNDOC
select
    REPLACE(REPLACE(REPLACE(data_row.`PK_PSNDOC`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNDOC`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_CORP`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_CORP`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_PSNCL`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNCL`,
    REPLACE(REPLACE(REPLACE(data_row.`PSNCODE`,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNCODE`,
    REPLACE(REPLACE(REPLACE(data_row.`PSNNAME`,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNNAME`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_DEPTDOC`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_DEPTDOC`,
    REPLACE(REPLACE(REPLACE(data_row.`CLERKFLAG`,CHR(10),''),CHR(13),''),CHR(9),'') as `CLERKFLAG`,
    REPLACE(REPLACE(REPLACE(data_row.`CLERKCODE`,CHR(10),''),CHR(13),''),CHR(9),'') as `CLERKCODE`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_OM_JOB`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_OM_JOB`,
    REPLACE(REPLACE(REPLACE(data_row.`INDOCFLAG`,CHR(10),''),CHR(13),''),CHR(9),'') as `INDOCFLAG`,
    CAST(REPLACE(REPLACE(REPLACE(data_row.`PSNCLSCOPE`,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `PSNCLSCOPE`,
    REPLACE(REPLACE(REPLACE(data_row.`AMCODE`,CHR(10),''),CHR(13),''),CHR(9),'') as `AMCODE`,
    REPLACE(REPLACE(REPLACE(data_row.`TS`,CHR(10),''),CHR(13),''),CHR(9),'') as `TS`,
    CAST(REPLACE(REPLACE(REPLACE(data_row.`DR`,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `DR`,
    REPLACE(REPLACE(REPLACE(data_row.`SEALDATE`,CHR(10),''),CHR(13),''),CHR(9),'') as `SEALDATE`,
    REPLACE(REPLACE(REPLACE(data_row.`INNERCODE`,CHR(10),''),CHR(13),''),CHR(9),'') as `INNERCODE`,
    REPLACE(REPLACE(REPLACE(data_row.`MAXINNERCODE`,CHR(10),''),CHR(13),''),CHR(9),'') as `MAXINNERCODE`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF1`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF1`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF10`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF10`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF11`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF11`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF12`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF12`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF13`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF13`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF14`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF14`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF15`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF15`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF16`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF16`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF17`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF17`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF18`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF18`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF19`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF19`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF2`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF2`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF20`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF20`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF3`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF3`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF4`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF4`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF5`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF5`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF6`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF6`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF7`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF7`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF8`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF8`,
    REPLACE(REPLACE(REPLACE(data_row.`DEF9`,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF9`,
    REPLACE(REPLACE(REPLACE(data_row.`INDUTYDATE`,CHR(10),''),CHR(13),''),CHR(9),'') as `INDUTYDATE`,
    REPLACE(REPLACE(REPLACE(data_row.`JOBRANK`,CHR(10),''),CHR(13),''),CHR(9),'') as `JOBRANK`,
    REPLACE(REPLACE(REPLACE(data_row.`JOBSERIES`,CHR(10),''),CHR(13),''),CHR(9),'') as `JOBSERIES`,
    REPLACE(REPLACE(REPLACE(data_row.`OUTDUTYDATE`,CHR(10),''),CHR(13),''),CHR(9),'') as `OUTDUTYDATE`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_PSNBASDOC`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNBASDOC`,
    REPLACE(REPLACE(REPLACE(data_row.`DIRECTLEADER`,CHR(10),''),CHR(13),''),CHR(9),'') as `DIRECTLEADER`,
    REPLACE(REPLACE(REPLACE(data_row.`DUTYNAME`,CHR(10),''),CHR(13),''),CHR(9),'') as `DUTYNAME`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF1`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF1`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF10`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF10`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF11`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF11`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF12`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF12`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF13`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF13`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF14`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF14`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF15`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF15`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF16`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF16`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF17`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF17`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF18`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF18`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF19`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF19`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF2`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF2`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF20`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF20`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF3`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF3`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF4`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF4`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF5`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF5`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF6`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF6`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF7`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF7`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF8`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF8`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF9`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF9`,
    REPLACE(REPLACE(REPLACE(data_row.`INSOURCE`,CHR(10),''),CHR(13),''),CHR(9),'') as `INSOURCE`,
    REPLACE(REPLACE(REPLACE(data_row.`ISCALOVERTIME`,CHR(10),''),CHR(13),''),CHR(9),'') as `ISCALOVERTIME`,
    REPLACE(REPLACE(REPLACE(data_row.`OUTMETHOD`,CHR(10),''),CHR(13),''),CHR(9),'') as `OUTMETHOD`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_CLERKCLASS`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_CLERKCLASS`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_DUTYRANK`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_DUTYRANK`,
    REPLACE(REPLACE(REPLACE(data_row.`PK_PSNTYPE`,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNTYPE`,
    REPLACE(REPLACE(REPLACE(data_row.`POSTSTAT`,CHR(10),''),CHR(13),''),CHR(9),'') as `POSTSTAT`,
    REPLACE(REPLACE(REPLACE(data_row.`RECRUITRESOURCE`,CHR(10),''),CHR(13),''),CHR(9),'') as `RECRUITRESOURCE`,
    REPLACE(REPLACE(REPLACE(data_row.`REGULAR`,CHR(10),''),CHR(13),''),CHR(9),'') as `REGULAR`,
    REPLACE(REPLACE(REPLACE(data_row.`REGULARDATA`,CHR(10),''),CHR(13),''),CHR(9),'') as `REGULARDATA`,
    REPLACE(REPLACE(REPLACE(data_row.`SERIES`,CHR(10),''),CHR(13),''),CHR(9),'') as `SERIES`,
    CAST(REPLACE(REPLACE(REPLACE(data_row.`SHOWORDER`,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `SHOWORDER`,
    CAST(REPLACE(REPLACE(REPLACE(data_row.`TBM_PROP`,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `TBM_PROP`,
    REPLACE(REPLACE(REPLACE(data_row.`TIMECARDID`,CHR(10),''),CHR(13),''),CHR(9),'') as `TIMECARDID`,
    REPLACE(REPLACE(REPLACE(data_row.`WASTOPDATE`,CHR(10),''),CHR(13),''),CHR(9),'') as `WASTOPDATE`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF1`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF1`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF2`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF2`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF3`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF3`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF4`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF4`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF5`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF5`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF6`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF6`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF7`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF7`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF8`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF8`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF9`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF9`,
    REPLACE(REPLACE(REPLACE(data_row.`CORPDEF10`,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF10`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF21`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF21`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF22`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF22`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF23`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF23`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF24`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF24`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF25`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF25`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF26`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF26`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF27`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF27`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF28`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF28`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF29`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF29`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF30`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF30`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF31`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF31`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF32`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF32`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF33`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF33`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF34`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF34`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF35`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF35`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF36`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF36`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF37`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF37`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF38`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF38`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF39`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF39`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF40`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF40`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF41`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF41`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF42`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF42`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF43`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF43`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF44`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF44`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF45`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF45`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF46`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF46`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF47`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF47`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF48`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF48`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF49`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF49`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF50`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF50`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF51`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF51`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF52`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF52`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF53`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF53`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF54`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF54`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF55`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF55`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF56`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF56`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF57`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF57`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF58`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF58`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF59`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF59`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF60`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF60`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF61`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF61`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF62`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF62`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF63`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF63`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF64`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF64`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF65`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF65`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF66`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF66`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF67`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF67`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF68`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF68`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF69`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF69`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF70`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF70`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF71`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF71`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF72`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF72`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF73`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF73`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF74`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF74`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF75`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF75`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF76`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF76`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF77`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF77`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF78`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF78`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF79`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF79`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF80`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF80`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF81`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF81`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF82`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF82`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF83`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF83`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF84`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF84`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF85`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF85`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF86`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF86`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF87`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF87`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF88`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF88`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF89`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF89`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF90`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF90`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF91`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF91`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF92`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF92`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF93`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF93`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF94`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF94`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF95`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF95`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF96`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF96`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF97`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF97`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF98`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF98`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF99`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF99`,
    REPLACE(REPLACE(REPLACE(data_row.`GROUPDEF100`,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF100`,
    REPLACE(REPLACE(REPLACE(data_row.`ISRETURN`,CHR(10),''),CHR(13),''),CHR(9),'') as `ISRETURN`,
    REPLACE(REPLACE(REPLACE(data_row.`ISREFERENCED`,CHR(10),''),CHR(13),''),CHR(9),'') as `ISREFERENCED`,
    REPLACE(REPLACE(REPLACE(data_row.`ONPOSTDATE`,CHR(10),''),CHR(13),''),CHR(9),'') as `ONPOSTDATE`,
    REPLACE(REPLACE(REPLACE(data_row.`CREATETIME`,CHR(10),''),CHR(13),''),CHR(9),'') as `CREATETIME`,
    REPLACE(REPLACE(REPLACE(data_row.`CREATOR`,CHR(10),''),CHR(13),''),CHR(9),'') as `CREATOR`,
    REPLACE(REPLACE(REPLACE(data_row.`MODIFIER`,CHR(10),''),CHR(13),''),CHR(9),'') as `MODIFIER`,
    REPLACE(REPLACE(REPLACE(data_row.`MODIFYTIME`,CHR(10),''),CHR(13),''),CHR(9),'') as `MODIFYTIME`,
    REPLACE(REPLACE(REPLACE(data_row.`HROPERATOR`,CHR(10),''),CHR(13),''),CHR(9),'') as `HROPERATOR`,
    REPLACE(REPLACE(REPLACE(data_row.`PSNNAMEPINYIN`,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNNAMEPINYIN`,
    `op_type` as `OP_TYPE`,
    `op_ts` as `OP_TS`,
    `current_ts` as `OGG_TIME`,
    CURRENT_TIMESTAMP as `GMT_CREATE`
from KAFKA_SOURCE_FIN_NC_BD_PSNDOC;

一直报错，经研究发现不能使用datarow的写法，需要改写成case when

-- 第二批
CREATE TABLE KAFKA_SOURCE_FIN_NC_BD_PSNDOC (
    op_type String,
    op_ts String,
    op_ts_time as cast(op_ts as timestamp(3)),
    current_ts String,
    pos String,
    before row<`PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` String,
    `AMCODE` String,
    `TS` String,
    `DR` String,
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` String,
    `TBM_PROP` String,
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String >,
    after row<`PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` String,
    `AMCODE` String,
    `TS` String,
    `DR` String,
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` String,
    `TBM_PROP` String,
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String >,
    data_row as
    case
    when op_type = 'D' then before
    else after
    end,
    watermark for op_ts_time as op_ts_time - interval '5' second
) with (
    'connector' = 'kafka',
    'topic' = 'SEM.NC.JXNC57.BD_PSNDOC',
    'scan.startup.mode' = 'earliest-offset',
    'properties.group.id' = 'pipeline_kafka2doris_3_test',
    'properties.bootstrap.servers' = 'jxbigdatakafka01.juneyaoair.com:9092,jxbigdatakafka02.juneyaoair.com:9092,jxbigdatakafka03.juneyaoair.com:9092',
    'properties.fetch.max.bytes' = '5242880',
    'properties.allow.auto.create.topics' = 'false',
    'properties.enable.auto.commit' = 'true',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username="bigdatauser" password="bigdata@gd66";',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'format' = 'json'
);

CREATE TABLE DORIS_SINK_FIN_NC_BD_PSNDOC (
    `PK_PSNDOC` String,
    `PK_CORP` String,
    `PK_PSNCL` String,
    `PSNCODE` String,
    `PSNNAME` String,
    `PK_DEPTDOC` String,
    `CLERKFLAG` String,
    `CLERKCODE` String,
    `PK_OM_JOB` String,
    `INDOCFLAG` String,
    `PSNCLSCOPE` DECIMAL,
    `AMCODE` String,
    `TS` String,
    `DR` DECIMAL(10,0),
    `SEALDATE` String,
    `INNERCODE` String,
    `MAXINNERCODE` String,
    `DEF1` String,
    `DEF10` String,
    `DEF11` String,
    `DEF12` String,
    `DEF13` String,
    `DEF14` String,
    `DEF15` String,
    `DEF16` String,
    `DEF17` String,
    `DEF18` String,
    `DEF19` String,
    `DEF2` String,
    `DEF20` String,
    `DEF3` String,
    `DEF4` String,
    `DEF5` String,
    `DEF6` String,
    `DEF7` String,
    `DEF8` String,
    `DEF9` String,
    `INDUTYDATE` String,
    `JOBRANK` String,
    `JOBSERIES` String,
    `OUTDUTYDATE` String,
    `PK_PSNBASDOC` String,
    `DIRECTLEADER` String,
    `DUTYNAME` String,
    `GROUPDEF1` String,
    `GROUPDEF10` String,
    `GROUPDEF11` String,
    `GROUPDEF12` String,
    `GROUPDEF13` String,
    `GROUPDEF14` String,
    `GROUPDEF15` String,
    `GROUPDEF16` String,
    `GROUPDEF17` String,
    `GROUPDEF18` String,
    `GROUPDEF19` String,
    `GROUPDEF2` String,
    `GROUPDEF20` String,
    `GROUPDEF3` String,
    `GROUPDEF4` String,
    `GROUPDEF5` String,
    `GROUPDEF6` String,
    `GROUPDEF7` String,
    `GROUPDEF8` String,
    `GROUPDEF9` String,
    `INSOURCE` String,
    `ISCALOVERTIME` String,
    `OUTMETHOD` String,
    `PK_CLERKCLASS` String,
    `PK_DUTYRANK` String,
    `PK_PSNTYPE` String,
    `POSTSTAT` String,
    `RECRUITRESOURCE` String,
    `REGULAR` String,
    `REGULARDATA` String,
    `SERIES` String,
    `SHOWORDER` DECIMAL(38,0),
    `TBM_PROP` DECIMAL(38,0),
    `TIMECARDID` String,
    `WASTOPDATE` String,
    `CORPDEF1` String,
    `CORPDEF2` String,
    `CORPDEF3` String,
    `CORPDEF4` String,
    `CORPDEF5` String,
    `CORPDEF6` String,
    `CORPDEF7` String,
    `CORPDEF8` String,
    `CORPDEF9` String,
    `CORPDEF10` String,
    `GROUPDEF21` String,
    `GROUPDEF22` String,
    `GROUPDEF23` String,
    `GROUPDEF24` String,
    `GROUPDEF25` String,
    `GROUPDEF26` String,
    `GROUPDEF27` String,
    `GROUPDEF28` String,
    `GROUPDEF29` String,
    `GROUPDEF30` String,
    `GROUPDEF31` String,
    `GROUPDEF32` String,
    `GROUPDEF33` String,
    `GROUPDEF34` String,
    `GROUPDEF35` String,
    `GROUPDEF36` String,
    `GROUPDEF37` String,
    `GROUPDEF38` String,
    `GROUPDEF39` String,
    `GROUPDEF40` String,
    `GROUPDEF41` String,
    `GROUPDEF42` String,
    `GROUPDEF43` String,
    `GROUPDEF44` String,
    `GROUPDEF45` String,
    `GROUPDEF46` String,
    `GROUPDEF47` String,
    `GROUPDEF48` String,
    `GROUPDEF49` String,
    `GROUPDEF50` String,
    `GROUPDEF51` String,
    `GROUPDEF52` String,
    `GROUPDEF53` String,
    `GROUPDEF54` String,
    `GROUPDEF55` String,
    `GROUPDEF56` String,
    `GROUPDEF57` String,
    `GROUPDEF58` String,
    `GROUPDEF59` String,
    `GROUPDEF60` String,
    `GROUPDEF61` String,
    `GROUPDEF62` String,
    `GROUPDEF63` String,
    `GROUPDEF64` String,
    `GROUPDEF65` String,
    `GROUPDEF66` String,
    `GROUPDEF67` String,
    `GROUPDEF68` String,
    `GROUPDEF69` String,
    `GROUPDEF70` String,
    `GROUPDEF71` String,
    `GROUPDEF72` String,
    `GROUPDEF73` String,
    `GROUPDEF74` String,
    `GROUPDEF75` String,
    `GROUPDEF76` String,
    `GROUPDEF77` String,
    `GROUPDEF78` String,
    `GROUPDEF79` String,
    `GROUPDEF80` String,
    `GROUPDEF81` String,
    `GROUPDEF82` String,
    `GROUPDEF83` String,
    `GROUPDEF84` String,
    `GROUPDEF85` String,
    `GROUPDEF86` String,
    `GROUPDEF87` String,
    `GROUPDEF88` String,
    `GROUPDEF89` String,
    `GROUPDEF90` String,
    `GROUPDEF91` String,
    `GROUPDEF92` String,
    `GROUPDEF93` String,
    `GROUPDEF94` String,
    `GROUPDEF95` String,
    `GROUPDEF96` String,
    `GROUPDEF97` String,
    `GROUPDEF98` String,
    `GROUPDEF99` String,
    `GROUPDEF100` String,
    `ISRETURN` String,
    `ISREFERENCED` String,
    `ONPOSTDATE` String,
    `CREATETIME` String,
    `CREATOR` String,
    `MODIFIER` String,
    `MODIFYTIME` String,
    `HROPERATOR` String,
    `PSNNAMEPINYIN` String,
    `OP_TYPE` String,
    `OP_TS` String,
    `OGG_TIME` String,
    `GMT_CREATE` TIMESTAMP,
    PRIMARY KEY (`PK_PSNDOC`) NOT ENFORCED
) with (
    'connector' = 'doris',
    'fenodes' = '10.17.23.55:8030',
    'table.identifier' = 'ods.ods_fin_nc_bd_psndoc',
    'username' = 'juneyaoair_etl',
    'password' = 'juneyaoair_etl123!'
);

insert into DORIS_SINK_FIN_NC_BD_PSNDOC
select
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_PSNDOC` else after.`PK_PSNDOC` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNDOC`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_CORP` else after.`PK_CORP` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_CORP`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_PSNCL` else after.`PK_PSNCL` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNCL`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PSNCODE` else after.`PSNCODE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNCODE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PSNNAME` else after.`PSNNAME` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNNAME`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_DEPTDOC` else after.`PK_DEPTDOC` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_DEPTDOC`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CLERKFLAG` else after.`CLERKFLAG` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CLERKFLAG`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CLERKCODE` else after.`CLERKCODE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CLERKCODE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_OM_JOB` else after.`PK_OM_JOB` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_OM_JOB`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`INDOCFLAG` else after.`INDOCFLAG` end,CHR(10),''),CHR(13),''),CHR(9),'') as `INDOCFLAG`,
    CAST(REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PSNCLSCOPE` else after.`PSNCLSCOPE` end,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `PSNCLSCOPE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`AMCODE` else after.`AMCODE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `AMCODE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`TS` else after.`TS` end,CHR(10),''),CHR(13),''),CHR(9),'') as `TS`,
    CAST(REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DR` else after.`DR` end,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `DR`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`SEALDATE` else after.`SEALDATE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `SEALDATE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`INNERCODE` else after.`INNERCODE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `INNERCODE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`MAXINNERCODE` else after.`MAXINNERCODE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `MAXINNERCODE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF1` else after.`DEF1` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF1`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF10` else after.`DEF10` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF10`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF11` else after.`DEF11` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF11`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF12` else after.`DEF12` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF12`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF13` else after.`DEF13` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF13`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF14` else after.`DEF14` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF14`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF15` else after.`DEF15` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF15`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF16` else after.`DEF16` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF16`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF17` else after.`DEF17` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF17`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF18` else after.`DEF18` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF18`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF19` else after.`DEF19` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF19`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF2` else after.`DEF2` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF2`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF20` else after.`DEF20` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF20`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF3` else after.`DEF3` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF3`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF4` else after.`DEF4` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF4`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF5` else after.`DEF5` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF5`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF6` else after.`DEF6` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF6`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF7` else after.`DEF7` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF7`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF8` else after.`DEF8` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF8`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DEF9` else after.`DEF9` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DEF9`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`INDUTYDATE` else after.`INDUTYDATE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `INDUTYDATE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`JOBRANK` else after.`JOBRANK` end,CHR(10),''),CHR(13),''),CHR(9),'') as `JOBRANK`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`JOBSERIES` else after.`JOBSERIES` end,CHR(10),''),CHR(13),''),CHR(9),'') as `JOBSERIES`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`OUTDUTYDATE` else after.`OUTDUTYDATE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `OUTDUTYDATE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_PSNBASDOC` else after.`PK_PSNBASDOC` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNBASDOC`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DIRECTLEADER` else after.`DIRECTLEADER` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DIRECTLEADER`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`DUTYNAME` else after.`DUTYNAME` end,CHR(10),''),CHR(13),''),CHR(9),'') as `DUTYNAME`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF1` else after.`GROUPDEF1` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF1`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF10` else after.`GROUPDEF10` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF10`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF11` else after.`GROUPDEF11` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF11`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF12` else after.`GROUPDEF12` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF12`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF13` else after.`GROUPDEF13` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF13`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF14` else after.`GROUPDEF14` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF14`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF15` else after.`GROUPDEF15` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF15`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF16` else after.`GROUPDEF16` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF16`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF17` else after.`GROUPDEF17` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF17`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF18` else after.`GROUPDEF18` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF18`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF19` else after.`GROUPDEF19` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF19`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF2` else after.`GROUPDEF2` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF2`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF20` else after.`GROUPDEF20` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF20`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF3` else after.`GROUPDEF3` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF3`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF4` else after.`GROUPDEF4` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF4`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF5` else after.`GROUPDEF5` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF5`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF6` else after.`GROUPDEF6` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF6`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF7` else after.`GROUPDEF7` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF7`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF8` else after.`GROUPDEF8` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF8`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF9` else after.`GROUPDEF9` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF9`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`INSOURCE` else after.`INSOURCE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `INSOURCE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`ISCALOVERTIME` else after.`ISCALOVERTIME` end,CHR(10),''),CHR(13),''),CHR(9),'') as `ISCALOVERTIME`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`OUTMETHOD` else after.`OUTMETHOD` end,CHR(10),''),CHR(13),''),CHR(9),'') as `OUTMETHOD`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_CLERKCLASS` else after.`PK_CLERKCLASS` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_CLERKCLASS`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_DUTYRANK` else after.`PK_DUTYRANK` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_DUTYRANK`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PK_PSNTYPE` else after.`PK_PSNTYPE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PK_PSNTYPE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`POSTSTAT` else after.`POSTSTAT` end,CHR(10),''),CHR(13),''),CHR(9),'') as `POSTSTAT`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`RECRUITRESOURCE` else after.`RECRUITRESOURCE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `RECRUITRESOURCE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`REGULAR` else after.`REGULAR` end,CHR(10),''),CHR(13),''),CHR(9),'') as `REGULAR`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`REGULARDATA` else after.`REGULARDATA` end,CHR(10),''),CHR(13),''),CHR(9),'') as `REGULARDATA`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`SERIES` else after.`SERIES` end,CHR(10),''),CHR(13),''),CHR(9),'') as `SERIES`,
    CAST(REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`SHOWORDER` else after.`SHOWORDER` end,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `SHOWORDER`,
    CAST(REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`TBM_PROP` else after.`TBM_PROP` end,CHR(10),''),CHR(13),''),CHR(9),'') AS DECIMAL) as `TBM_PROP`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`TIMECARDID` else after.`TIMECARDID` end,CHR(10),''),CHR(13),''),CHR(9),'') as `TIMECARDID`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`WASTOPDATE` else after.`WASTOPDATE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `WASTOPDATE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF1` else after.`CORPDEF1` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF1`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF2` else after.`CORPDEF2` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF2`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF3` else after.`CORPDEF3` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF3`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF4` else after.`CORPDEF4` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF4`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF5` else after.`CORPDEF5` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF5`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF6` else after.`CORPDEF6` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF6`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF7` else after.`CORPDEF7` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF7`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF8` else after.`CORPDEF8` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF8`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF9` else after.`CORPDEF9` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF9`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CORPDEF10` else after.`CORPDEF10` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CORPDEF10`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF21` else after.`GROUPDEF21` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF21`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF22` else after.`GROUPDEF22` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF22`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF23` else after.`GROUPDEF23` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF23`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF24` else after.`GROUPDEF24` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF24`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF25` else after.`GROUPDEF25` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF25`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF26` else after.`GROUPDEF26` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF26`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF27` else after.`GROUPDEF27` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF27`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF28` else after.`GROUPDEF28` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF28`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF29` else after.`GROUPDEF29` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF29`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF30` else after.`GROUPDEF30` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF30`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF31` else after.`GROUPDEF31` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF31`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF32` else after.`GROUPDEF32` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF32`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF33` else after.`GROUPDEF33` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF33`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF34` else after.`GROUPDEF34` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF34`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF35` else after.`GROUPDEF35` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF35`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF36` else after.`GROUPDEF36` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF36`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF37` else after.`GROUPDEF37` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF37`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF38` else after.`GROUPDEF38` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF38`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF39` else after.`GROUPDEF39` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF39`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF40` else after.`GROUPDEF40` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF40`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF41` else after.`GROUPDEF41` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF41`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF42` else after.`GROUPDEF42` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF42`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF43` else after.`GROUPDEF43` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF43`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF44` else after.`GROUPDEF44` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF44`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF45` else after.`GROUPDEF45` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF45`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF46` else after.`GROUPDEF46` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF46`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF47` else after.`GROUPDEF47` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF47`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF48` else after.`GROUPDEF48` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF48`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF49` else after.`GROUPDEF49` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF49`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF50` else after.`GROUPDEF50` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF50`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF51` else after.`GROUPDEF51` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF51`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF52` else after.`GROUPDEF52` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF52`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF53` else after.`GROUPDEF53` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF53`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF54` else after.`GROUPDEF54` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF54`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF55` else after.`GROUPDEF55` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF55`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF56` else after.`GROUPDEF56` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF56`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF57` else after.`GROUPDEF57` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF57`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF58` else after.`GROUPDEF58` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF58`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF59` else after.`GROUPDEF59` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF59`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF60` else after.`GROUPDEF60` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF60`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF61` else after.`GROUPDEF61` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF61`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF62` else after.`GROUPDEF62` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF62`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF63` else after.`GROUPDEF63` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF63`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF64` else after.`GROUPDEF64` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF64`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF65` else after.`GROUPDEF65` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF65`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF66` else after.`GROUPDEF66` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF66`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF67` else after.`GROUPDEF67` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF67`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF68` else after.`GROUPDEF68` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF68`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF69` else after.`GROUPDEF69` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF69`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF70` else after.`GROUPDEF70` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF70`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF71` else after.`GROUPDEF71` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF71`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF72` else after.`GROUPDEF72` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF72`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF73` else after.`GROUPDEF73` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF73`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF74` else after.`GROUPDEF74` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF74`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF75` else after.`GROUPDEF75` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF75`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF76` else after.`GROUPDEF76` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF76`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF77` else after.`GROUPDEF77` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF77`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF78` else after.`GROUPDEF78` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF78`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF79` else after.`GROUPDEF79` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF79`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF80` else after.`GROUPDEF80` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF80`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF81` else after.`GROUPDEF81` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF81`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF82` else after.`GROUPDEF82` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF82`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF83` else after.`GROUPDEF83` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF83`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF84` else after.`GROUPDEF84` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF84`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF85` else after.`GROUPDEF85` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF85`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF86` else after.`GROUPDEF86` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF86`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF87` else after.`GROUPDEF87` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF87`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF88` else after.`GROUPDEF88` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF88`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF89` else after.`GROUPDEF89` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF89`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF90` else after.`GROUPDEF90` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF90`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF91` else after.`GROUPDEF91` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF91`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF92` else after.`GROUPDEF92` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF92`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF93` else after.`GROUPDEF93` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF93`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF94` else after.`GROUPDEF94` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF94`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF95` else after.`GROUPDEF95` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF95`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF96` else after.`GROUPDEF96` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF96`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF97` else after.`GROUPDEF97` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF97`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF98` else after.`GROUPDEF98` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF98`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF99` else after.`GROUPDEF99` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF99`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`GROUPDEF100` else after.`GROUPDEF100` end,CHR(10),''),CHR(13),''),CHR(9),'') as `GROUPDEF100`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`ISRETURN` else after.`ISRETURN` end,CHR(10),''),CHR(13),''),CHR(9),'') as `ISRETURN`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`ISREFERENCED` else after.`ISREFERENCED` end,CHR(10),''),CHR(13),''),CHR(9),'') as `ISREFERENCED`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`ONPOSTDATE` else after.`ONPOSTDATE` end,CHR(10),''),CHR(13),''),CHR(9),'') as `ONPOSTDATE`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CREATETIME` else after.`CREATETIME` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CREATETIME`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`CREATOR` else after.`CREATOR` end,CHR(10),''),CHR(13),''),CHR(9),'') as `CREATOR`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`MODIFIER` else after.`MODIFIER` end,CHR(10),''),CHR(13),''),CHR(9),'') as `MODIFIER`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`MODIFYTIME` else after.`MODIFYTIME` end,CHR(10),''),CHR(13),''),CHR(9),'') as `MODIFYTIME`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`HROPERATOR` else after.`HROPERATOR` end,CHR(10),''),CHR(13),''),CHR(9),'') as `HROPERATOR`,
    REPLACE(REPLACE(REPLACE(case when op_type = 'D' then before.`PSNNAMEPINYIN` else after.`PSNNAMEPINYIN` end,CHR(10),''),CHR(13),''),CHR(9),'') as `PSNNAMEPINYIN`,
    `op_type` as `OP_TYPE`,
    `op_ts` as `OP_TS`,
    `current_ts` as `OGG_TIME`,
    CURRENT_TIMESTAMP as `GMT_CREATE`
from KAFKA_SOURCE_FIN_NC_BD_PSNDOC;