/dps

Data processing system for polyglot

Primary LanguagePython

DPS (Data Processing System)

Requriements

  • python 3.8

How to run DPS?

python setup.py install
python bin/sparkapp.py {job_name} {params}

# Example
# python bin/sparkapp.py sample_job --config_path=./configs/sample_job.yaml

DPS job list

job describe param options
sample_job Sample jsonl data from text files in directories yaml configs
dedup_job De-duplicate jsonl data using MinHash method yaml configs
korean_job Refine jsonl data in Korean language yaml configs

Development guides

Test Run

This is test run for sample_job job.

1. Setup dps package

python setup.py install

2. Check config file and dataset

cat configs/sample_job.yaml
ls datasets/test_sample_jsonl_data

3. Run sample_job job by bin/sparkapp.py

python bin/sparkapp.py sample_job --config_path=./configs/sample_job.yaml

4. Check output file

cat datasets/test_output_data/part-00000

Add your own job

Implement your job function

  1. Make an issue on ElutherAI/dps repository

    • Describe your job first
    • Define input and outputs and these examples
  2. Go to dps/spark/jobs and create python your_own_job.py script file.

  3. Make a function to run your job. Here's template to play your works.

    from pyspark import SparkContext
    from pyspark.rdd import RDD
    
    from dps.spark.spark_session import spark_session
    from dps.spark.utils.io_utils import read_line, to_json
    
    
    def your_own_job(input_path, output_path):
        
        with spark_session(f'your own job') as spark:
            sc: SparkContext = spark.sparkContext # Spark context is to run your spark application
    
            # Read all files in your directory or file
            proc_rdd: RDD = sc.textFile(input_path) \
                .repartition(10) \
                .flatMap(read_line) 
                
            # Write data that you processed
            proc_rdd \
                .repartition(1) \
                .flatMap(to_json) \
                .saveAsTextFile(output_path)
  4. Register your job into dps/spark/run.py

    from .jobs.your_own_job import your_own_job
    
    def run():
        fire.Fire({'sample_job': sample_job,
                   'your_own_job': your_own_job
                   })
  5. Test run your job

    python bin/sparkapp.py your_own_job --input_path='{input_your_data_dir_or_file}' \
                                        --output_path='{output_path}'