

def test_load_data_partition(self):
    df = self.df
    unpart_t = self.db.table(self.pd_name)
    part_keys = ['year', 'month']
    part_t = self._create_partitioned_table(unpart_t.schema(), part_keys)
    df = df[(df.month == '1')].reset_index(drop=True)
    unique_keys = df[part_keys].drop_duplicates()
    hdfs_dir = pjoin(self.tmp_dir, 'load-data-partition')
    df2 = df.drop(['year', 'month'], axis='columns')
    csv_props = {
        'serialization.format': ',',
        'field.delim': ',',
    }
    for (i, (year, month)) in enumerate(unique_keys.itertuples(index=False)):
        chunk = df2[((df.year == year) & (df.month == month))]
        chunk_path = pjoin(hdfs_dir, '{0}.csv'.format(i))
        self.con.write_dataframe(chunk, chunk_path)
        if i:
            part = {
                'year': year,
                'month': month,
            }
        else:
            part = [year, month]
        part_t.add_partition(part)
        part_t.alter_partition(part, format='text', serde_properties=csv_props)
        part_t.load_data(chunk_path, partition=part)
    self.hdfs.rmdir(hdfs_dir)
    self._verify_partitioned_table(part_t, df, unique_keys)

dataset/ETHPy150Open cloudera/ibis/ibis/impala/tests/test_partition.py TestPartitioning.test_load_data_partition/original