Hashing values using SHA-2 in PySpark

SHA stands for Secure Hashing Algorithm and 2 is just a version number. SHA-2 revises the construction and the big-length of the signature from SHA-1. You may also see SHA-224, SHA-256, SHA-384 or SHA-512. Those are referring the bit-lengths of SHA-2. It's a bit confusing. SHA-2 produces irreversible and unique hashes as it is a one-way hash function. The original data remains secure and unknown. However, given than a SHA-2 function with L message digest bits, you can have maximum 2^L possibilities, which means somebody can perform a brute force search even though it is not quite practical. Moreover, with a precomputed table, called rainbow table for caching the output of hash functions, simple input can be easily cracked. Therefore, to prevent precomputation attacks, we need salting. Salting is just an additional input concatenating to the original input. It should be long and random every time calling the hash function. ``` saltedhash(input) = hash_function(salt + input) ``` In PySpark, ``sha2`` was implemented since version 1.5. ``` def sha2(col, numBits): """Returns the hex string result of SHA-2 family of hash functions (SHA-224, SHA-256, SHA-384, and SHA-512). The numBits indicates the desired bit length of the result, which must have a value of 224, 256, 384, 512, or 0 (which is equivalent to 256). >>> digests = df.select(sha2(df.name, 256).alias('s')).collect() >>> digests[0] Row(s=u'3bc51062973c458d5a6f2d8d64a023246354ad7e064b1e4e009ec8a0699a3043') >>> digests[1] Row(s=u'cd9fb1e148ccd8442e5aa74904cc73bf6fb54d1d54d333bd596aa9bb4bb4e961') """ sc = SparkContext._active_spark_context jc = sc._jvm.functions.sha2(_to_java_column(col), numBits) return Column(jc) ``` First, we need to import the functions. ``` from pyspark.sql.functions import concat, col, lit, bin, sha2 ``` This is an example using ``withColumn`` with ``sha2`` function to hash the salt and the input with 256 message digest bits. ``` df = df.withColumn( col_name, sha2(concat(lit(generate_salt()), bin(col(col_name))), 256) ) ``` The hash value looks like ``8ba06918c277ee2e9b6eecb798fe64dc4a8c34d95b4514ecc267487aee9b84b9``.

Refactoring messy Test Suite in Python

Recently, I did some code reviews on my peer's code. We've created 100+ pySpark jobs and each job has its own test cases. When I looked at the file, it was super long. Let's take a look at main function. It is pretty general test suite main function. ``` if __name__ == "__main__": loader = unittest.TestLoader() suite = create_unit_suite() runner = unittest.TextTestRunner(verbosity=2) runner.run(suite) ``` However, when diving into create_unit_suite function, I was utterly dumbfounded. ``` def create_unit_suite(): suite = unittest.TestSuite() suite.addTests(loader.loadTestsFromModule(tests.test_job_1)) suite.addTests(loader.loadTestsFromModule(tests.test_job_2)) suite.addTests(loader.loadTestsFromModule(tests.test_job_3)) # ... suite.addTests(loader.loadTestsFromModule(tests.test_job_100)) suite.addTests(loader.loadTestsFromModule(tests.test_job_101)) suite.addTests(loader.loadTestsFromModule(tests.test_job_102)) # ... ``` And the import statements look like ``` import tests.test_job_1 import tests.test_job_2 import tests.test_job_3 # ... import tests.test_job_100 import tests.test_job_101 import tests.test_job_102 # ... ``` Every time a new job is created, the developers need to import it and and add a new addTests line. The folder structure is way more complicated and there is no guarantee that all developers remember to add the tests. You can't tell which jobs are missing test cases simply by looking at this file. Thanks to unittest, there is a function called ``discovery`` introduced in v3.2. It can be used like TestLoader.discover() or from the command line. From the documentation, it states that Unittest supports simple test discovery. In order to be compatible with test discovery, all of the test files must be modules or packages (including namespace packages) importable from the top-level directory of the project (this means that their filenames must be valid identifiers). In order to expose the test files, we need to make them modules. We can do that simply by adding __init__.py. Here's what's left. Specify the path and pattern for unittest to discover. No long import statements and addTests code. ``` import unittest from tests import * if __name__ == "__main__": testsuite = unittest.TestLoader().discover("tests", pattern="test_*.py") runner = unittest.TextTestRunner(verbosity=2).run(testsuite) ``` Simply run the below command to get the result. ``` python -m unittest ``` However, running all the test cases may take a great of time. In order to have more flexibility, it should be able to run in a specific module. We can use ``argparse`` to take ``--module`` flag to determine which module we are going to run test cases on. ``` import argparse ``` Put the logic in main() ``` def main(): parser = argparse.ArgumentParser() parser.add_argument("--module", action="store", default="all", help="the module of a collection of test cases to be executed") args = parser.parse_args() target_module = args.module target_path = "tests" if target_module == "all" else "tests/{}".format(target_module) testsuite = unittest.TestLoader().discover(target_path, pattern="test_*.py") runner = unittest.TextTestRunner(verbosity=2).run(testsuite) if __name__ == "__main__": main() ```

Removing duplicate records in a CSV file

Recently my colleague was struggling on removing duplicate records based on composite keys in a large csv file before inserting into Oracle database. I wrote a simple python program to address the problem. Given that python has already been installed in the server. I could install ``pandas`` in virtualenv using pip. The program is pretty short. First we need to import pandas. Pandas is a software library for data manipulation and analysis. ```python import pandas as pd ``` Then, we use ``read_csv()`` to read a csv file into DataFrame, which is a 2d size-mutatable, potentially heterogeneous tabular data structure with labeled axes. Since our data contains a value of NA, so we need to set `keep_default_na` to False. ```python d = pd.read_csv('LARGE_CSV_FILE.csv', keep_default_na = False) ``` Once we have the dataframe, we can call ``drop_duplicates()`` to remove duplicate rows. Since we remove them based on composite keys, we can pass those keys to `subset`. Setting `inplace` to True can drop duplicates in place instead of returning a copy. If depulicate records are found, we only keep the first one. ```python d.drop_duplicates(subset = ['COMPOSITE_KEY1', 'COMPOSITE_KEY2', 'COMPOSITE_KEY3', 'COMPOSITE_KEY4', 'COMPOSITE_KEY5', 'COMPOSITE_KEY6', 'COMPOSITE_KEY7', 'COMPOSITE_KEY8', 'COMPOSITE_KEY9', 'COMPOSITE_KEY10'], inplace = True, keep = 'first') ``` At the end, we would like to save the result to another csv file for verification. By default, it comes with index. We can disable it by setting `index` to False. ```python d.to_csv('LARGE_CSV_FILE_PROCESSED.csv', index = False) ```