Tensorflow Data Validation 사용하기


  • TFX에서 Data Validation을 담당하고 있는 TensorFlow Data Validation(TFDV)에 대한 글입니다

TFDV

  • Github Repo
  • TensorFlow Data Validation는 데이터를 더 쉽게 이해하고 점검할 수 있도록 도와주는 라이브러리
  • Google에서 매일 페타바이트의 데이터를 분석하고 점검할 때 사용
  • Train, Test Data의 요약 통계, 분포를 쉽게 비교 가능
  • 필수 값, 데이터에 대한 기대치를 의미하는 데이터 스키마 생성
  • 지정된 조건을 벗어나는 경우(예시 : int 타입을 예상했는데 float이 들어오는 경우) 예외를 식별
  • Python 2.7, 3.5만 지원
  • Mac OS(시에라) 이상, Ubuntu 16.04 이상 지원
    • 윈도우는 추후 지원 예정
  • 내부적으로 Facet, Apache Beam 사용
  • 회사 업무시, 데이터 경진대회(캐글 등)에서 유용할 듯
  • 이 글은 0.13.1 버전 기준으로 작성했습니다. 추후 계속 수정될 예정입니다
  • 코드는 모두 Github에 올렸습니다. Nbviewer로 확인 부탁드려요!

기능

  • Statistics 생성 및 시각화
    • Train과 Test셋 데이터 분포 동시에 확인
  • 스키마 추론
  • 데이터 검증
  • Drift와 Skew 체크

설치

pip install tensorflow-data-validation
  • 설치한 후, 확인

      import sys, os
      import tempfile, urllib, zipfile
      import tensorflow_data_validation as tfdv
      print('TFDV version: {}'.format(tfdv.version.__version__))
    

데이터 준비

  • 시카고 택시 데이터를 다운로드 후, 압축 해제

      BASE_DIR = tempfile.mkdtemp()
      DATA_DIR = os.path.join(BASE_DIR, 'data')
      OUTPUT_DIR = os.path.join(BASE_DIR, 'chicago_taxi_output')
      TRAIN_DATA = os.path.join(DATA_DIR, 'train', 'data.csv')
      EVAL_DATA = os.path.join(DATA_DIR, 'eval', 'data.csv')
      SERVING_DATA = os.path.join(DATA_DIR, 'serving', 'data.csv')
    	
      zip, headers = urllib.urlretrieve('https://storage.googleapis.com/tfx-colab-datasets/chicago_data.zip')
      zipfile.ZipFile(zip).extractall(BASE_DIR)
      zipfile.ZipFile(zip).close()
    

Statistics 생성 및 시각화

  • csv, dataframe, tfrecord을 통해 Statistics를 생성할 수 있음

      [methods for methods in dir(tfdv) if "generate" in methods]
      >>> ['generate_statistics_from_csv',
       	 'generate_statistics_from_dataframe',
           'generate_statistics_from_tfrecord']
    
  • Statistics 생성

      train_stats = tfdv.generate_statistics_from_csv(data_location=TRAIN_DATA)
      # method : tfdv.generate_statistics_from_csv(data_location, column_names=None, delimiter=',', output_path=None, stats_options=<tensorflow_data_validation.statistics.stats_options.StatsOptions object at 0x14436a110>, pipeline_options=None)
    
    • train_stats를 확인해보면 feature별로 다양한 통계치를 가지고 있음
  • 시각화

      tfdv.visualize_statistics(train_stats)
      # method : tfdv.visualize_statistics(lhs_statistics, rhs_statistics=None, lhs_name='lhs_statistics', rhs_name='rhs_statistics)
    

    • Interactive하게 직접 만져보고 싶으신 분은 웹에서 제 Github nbviewer 참고!
    • Numeric Features
      • 맨 왼쪽에 Sort By 옵션
        • Non-uniformity
        • Amount missing/zero
      • 오른쪽에 Chart to show은 분포, Quantiles, Value list length 등으로 볼 수 있고, 클릭으로 log로 변환 가능
    • Categorical Feature도 아래처럼 시각화됨
      • SHOW RAW DATA를 누르면 카테고리 데이터 count 결과 보여줌
  • Train과 Eval Data 동시에 시각화

    • Kaggle에서 매우 유용할 듯
      eval_stats = tfdv.generate_statistics_from_csv(data_location=EVAL_DATA)
      tfdv.visualize_statistics(lhs_statistics=eval_stats, rhs_statistics=train_stats,
                            lhs_name='EVAL_DATASET', rhs_name='TRAIN_DATASET')
    

    • 이제 파란색은 Eval data, 주황색은 Train data
    • 우측 Charts to show에서 percentages를 클릭시 Eval과 Train의 비율을 포개서 보여줌

스키마 추론

  • tfdv.infer_schema로 스키마를 추론한 후, tfdv.display_shcema로 출력
  • infer schema 결과 스키마 Protocol buffer가 생성됨

      schema = tfdv.infer_schema(statistics=train_stats)
      # method : tfdv.infer_schema(statistics, infer_feature_shape=True, max_string_domain_size=100)
    	
      tfdv.display_schema(schema=schema)
    

    • 각 Feature별 Type, Presence, Valency, Domain 출력
  • Feautre의 속성을 바꾸고 싶을 경우 tfdv.get_feature로 가져온 후 수정

      tfdv.get_feature(schema, 'payment_type').name = "oh"
      schema 
      # find name: 'oh'
    
    • 다시 복구
      tfdv.get_feature(schema, 'oh').name = "payment_type"
    

데이터 검증

  • Validation 기능
  • 예를 들어 Train엔 없는 Categorical Value가 Test에 존재한다면?

      anomalies = tfdv.validate_statistics(statistics=eval_stats, schema=schema)
      tfdv.display_anomalies(anomalies)
    

    • company, payment type feature에 예상하지 못한 값이 있음
    • 문제 해결을 위해 min_domain_mass 제약 조건을 추가하고, value를 수동으로 추가함
      # Relax the minimum fraction of values that must come from the domain for feature company.
      company = tfdv.get_feature(schema, 'company')
      company.distribution_constraints.min_domain_mass = 0.9
    	
      # Add new value to the domain of feature payment_type.
      payment_type_domain = tfdv.get_domain(schema, 'payment_type')
      payment_type_domain.value.append('Prcard')
    	
      # Validate eval stats after updating the schema 
      updated_anomalies = tfdv.validate_statistics(eval_stats, schema)
      tfdv.display_anomalies(updated_anomalies)
    
  • Serving시 데이터 검증
    • requirements를 정의해 특정 문제가 발생하면 자동으로 수정될 수 있게 보조해주는 기능도 존재
      serving_stats = tfdv.generate_statistics_from_csv(SERVING_DATA)
      serving_anomalies = tfdv.validate_statistics(serving_stats, schema)
    
      tfdv.display_anomalies(serving_anomalies)
    

    • tips 컬럼은 완벽하게 missing이고 trip_seconds는 FLOAT 타입을 예상했는데 INT 타입이 들어옴
    • 타입 문제 해결 방법
      • infer_type_from_schema=True인 StatsOptions를 생성한 후, generate_statistics_from_csv에 추가
        options = tfdv.StatsOptions(schema=schema, infer_type_from_schema=True)
        serving_stats = tfdv.generate_statistics_from_csv(SERVING_DATA, stats_options=options)
        serving_anomalies = tfdv.validate_statistics(serving_stats, schema)
      
        tfdv.display_anomalies(serving_anomalies)
      
    • tips feature가 serving 데이터에 없는 문제 해결 방법
      • 스키마 환경을 TRAINING과 SERVING으로 생성한 후, SERVING시 tips 데이터가 없는 것을 명시
        # All features are by default in both TRAINING and SERVING environments.
        schema.default_environment.append('TRAINING')
        schema.default_environment.append('SERVING')
      		
        # Specify that 'tips' feature is not in SERVING environment.
        tfdv.get_feature(schema, 'tips').not_in_environment.append('SERVING')
      		
        serving_anomalies_with_env = tfdv.validate_statistics(
            serving_stats, schema, environment='SERVING')
      		
        tfdv.display_anomalies(serving_anomalies_with_env)
      

Drift와 Skew 체크

  • tfdv.validate_statistics() method를 사용해 체크
  • Drift
    • Drift detection은 Categorical 데이터 및 데이터의 연속 기간(N, N+1) 사이(예를 들면 서로 다른 날의 훈련 데이터 사이)에서 지원
    • L-infinity distnace로 Drift를 표현하고 허용값보다 높으면 경고를 받을 수 있음
    • 정확한 거리를 설정하는 것은 도메인 지식과 실험을 필요로하는 반복 프로세스
  • Skew
    • Schema Skew
      • 같은 스키마를 가지지 않을 때
    • Feature Skew
      • Feature 생성 로직이 변경될 때
    • Distribution Skew
      • Train, Serving 데이터 분포가 다를 경우
      # Add skew comparator for 'payment_type' feature.
      payment_type = tfdv.get_feature(schema, 'payment_type')
      payment_type.skew_comparator.infinity_norm.threshold = 0.01
    	
      # Add drift comparator for 'company' feature.
      company=tfdv.get_feature(schema, 'company')
      company.drift_comparator.infinity_norm.threshold = 0.001
    	
      skew_anomalies = tfdv.validate_statistics(train_stats, schema,
                                                previous_statistics=eval_stats,
                                                serving_statistics=serving_stats)
    	
      tfdv.display_anomalies(skew_anomalies)
    

  • payment_type이 train / serving시 거리가 매우 큼
  • company가 과거와 현재의 거리가 매우 큼
  • 위 예시에선 의도적으로 threshold값을 낮게 설정해 오류가 발생하도록 한 것이기 때문에 따로 수정하지 않음

스키마 저장

  • 스키마를 검토하고 큐레이션이 되었으니 “frozen”(고정된) 상태를 반영하도록 Protocol Buffer에 저장

      from tensorflow.python.lib.io import file_io
      from google.protobuf import text_format
    	
      file_io.recursive_create_dir(OUTPUT_DIR)
      schema_file = os.path.join(OUTPUT_DIR, 'schema.pbtxt')
      tfdv.write_schema_text(schema, schema_file)
    	
      !cat {schema_file}
    

데이터 형태가 다를 경우

  • 현재 TFRecord, CSV, Dataframe만 지원하는데 그 외의 데이터 타입을 사용한다면 Apache Beam의 PTransform을 사용해 데이터를 가공할 수 있음
  • 링크 참고하면 예시가 나와있음

구글 클라우드에서 사용하기

  • 설치

      pip download tensorflow_data_validation \
      --no-deps \
      --platform manylinux1_x86_64 \
      --only-binary=:all:
    
  • 예시 코드 snippet

      import tensorflow_data_validation as tfdv
      from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions
    	
      PROJECT_ID = ''
      JOB_NAME = ''
      GCS_STAGING_LOCATION = ''
      GCS_TMP_LOCATION = ''
      GCS_DATA_LOCATION = ''
      # GCS_STATS_OUTPUT_PATH is the file path to which to output the data statistics
      # result.
      GCS_STATS_OUTPUT_PATH = ''
    	
      PATH_TO_WHL_FILE = ''
    	
    	
      # Create and set your PipelineOptions.
      options = PipelineOptions()
    	
      # For Cloud execution, set the Cloud Platform project, job_name,
      # staging location, temp_location and specify DataflowRunner.
      google_cloud_options = options.view_as(GoogleCloudOptions)
      google_cloud_options.project = PROJECT_ID
      google_cloud_options.job_name = JOB_NAME
      google_cloud_options.staging_location = GCS_STAGING_LOCATION
      google_cloud_options.temp_location = GCS_TMP_LOCATION
      options.view_as(StandardOptions).runner = 'DataflowRunner'
    	
      setup_options = options.view_as(SetupOptions)
      # PATH_TO_WHL_FILE should point to the downloaded tfdv wheel file.
      setup_options.extra_packages = [PATH_TO_WHL_FILE]
    	
      tfdv.generate_statistics_from_tfrecord(GCS_DATA_LOCATION,
                                             output_path=GCS_STATS_OUTPUT_PATH,
                                             pipeline_options=options)
    

언제 TFDV를 사용해야할까?

  • 갑자기 이상한 feature가 들어오는지 확인하고 싶을 경우
  • Decision surface에서 모델이 훈련 잘되었는지 확인하고 싶을 경우
  • feature engineering 실수 방지하고 싶을 경우

Facets

  • 사실 TFDV의 데이터 분포 보여주는 부분은 Facets으로 이루어져 있음
  • Facets은 현재 2가지 시각화를 제공하는데, 1) FACETS OVERVIEW가 위에서 TFDV가 보여준 시각화고 2) FACETS DIVE로 다양한 양의 데이터를 한번에 인터랙티브하게 시각화해줌
  • FACETS DIVE는 아래처럼 시각화됨
    • 데이터를 연령대별 직업별로 보여줌
    • Quick Draw 데이터셋을 인터렉티브하게 보려면 링크 참고
  • Colab에서 사용하는 예제는 Github에 나와있음
    • 데이터를 데이터프레임으로 불러온 후, to_json을 사용해 json으로 변경
    • 그 후 HTML Template을 사용해 노트북에서 볼 수 있도록 출력
      # Load UCI census train and test data into dataframes.
      import pandas as pd
      features = ["Age", "Workclass", "fnlwgt", "Education", "Education-Num", "Marital Status",
                  "Occupation", "Relationship", "Race", "Sex", "Capital Gain", "Capital Loss",
                  "Hours per week", "Country", "Target"]
      train_data = pd.read_csv(
          "https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.data",
          names=features,
          sep=r'\s*,\s*',
          engine='python',
          na_values="?")
      test_data = pd.read_csv(
          "https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.test",
          names=features,
          sep=r'\s*,\s*',
          skiprows=[0],
          engine='python',
          na_values="?")
    	    
      # Display the Dive visualization for the training data.
      from IPython.core.display import display, HTML
    	
      jsonstr = train_data.to_json(orient='records')
      HTML_TEMPLATE = """
              <script src="https://cdnjs.cloudflare.com/ajax/libs/webcomponentsjs/0.7.24/webcomponents-lite.js"></script>
              <link rel="import" href="https://raw.githubusercontent.com/PAIR-code/facets/master/facets-dist/facets-jupyter.html">
              <facets-dive id="elem" height="600"></facets-dive>
              <script>
                var data = {jsonstr};
                document.querySelector("#elem").data = data;
              </script>"""
      html = HTML_TEMPLATE.format(jsonstr=jsonstr)
      display(HTML(html))
    

Refenrece


이 글이 도움이 되셨다면 추천 클릭을 부탁드립니다 :)

Buy me a coffeeBuy me a coffee





© 2017. by Seongyun Byeon

Powered by zzsza