def create_test_scalar_dataset(output_url, num_rows, num_files=4, spark=None, partition_by=None):
    """Creates a dataset in tmp_url location. The dataset emulates non-petastorm dataset, i.e. contains only native
    parquet types.

    These are the fields with mock data:
      'id', 'int_fixed_size_list', 'datetime', 'timestamp', 'string', 'string2', 'float64'

    :param output_url: Url specifying the location the parquet store is written to
    :param num_rows: Number of rows in the generated dataset
    :param num_files: Number of parquet files that will be written into the parquet store
    :param spark: An instance of spark session object. If `None` (default), a new spark session is created.
    :param partition_by: A list of fields to partition the parquet store by.
    :return: A list of records with a copy of the data written to the dataset.
    """

    partition_by = partition_by or []
    shutdown = False
    if not spark:
        spark_session = SparkSession \
            .builder \
            .appName('petastorm_end_to_end_test') \
            .master('local[*]')

        spark = spark_session.getOrCreate()
        shutdown = True
        hadoop_config = spark.sparkContext._jsc.hadoopConfiguration()
        hadoop_config.setInt('parquet.block.size', 100)

    def expected_row(i):
        result = {
            'id': np.int32(i),
            'id_div_700': np.int32(i // 700),
            'datetime': np.datetime64('2019-01-02'),
            'timestamp': np.datetime64('2005-02-25T03:30'),
            'string': np.unicode_('hello_{}'.format(i)),
            'string2': np.unicode_('world_{}'.format(i)),
            'float64': np.float64(i) * .66,
            'int_fixed_size_list': np.arange(1 + i, 10 + i).astype(np.int32)
        }
        return result

    expected_data = [expected_row(i) for i in range(num_rows)]

    expected_data_as_scalars = [{k: v.item() if isinstance(v, np.generic) else v for k, v in row.items()} for row
                                in expected_data]

    # np.datetime64 is converted to a timezone unaware datetime instances. Working explicitly in UTC so we don't need
    # to think about local timezone in the tests
    for row in expected_data_as_scalars:
        row['timestamp'] = row['timestamp'].replace(tzinfo=pytz.UTC)
        row['int_fixed_size_list'] = row['int_fixed_size_list'].tolist()
        row['nested_struct'] = {'nested_int': 0}

    rows = [Row(**OrderedDict(sorted(row.items(), key=lambda item: item[0]))) for row in expected_data_as_scalars]

    maybe_int_fixed_size_list_field = [StructField('int_fixed_size_list', ArrayType(IntegerType(), False), False)]

    # WARNING: surprisingly, schema fields and row fields are matched only by order and not name.
    # We must maintain alphabetical order of the struct fields for the code to work!!!
    schema = StructType(
        [
            StructField('datetime', DateType(), False),
            StructField('float64', DoubleType(), False),
            StructField('id', IntegerType(), False),
            StructField('id_div_700', IntegerType(), False),
        ] + maybe_int_fixed_size_list_field +
        [
            StructField('nested_struct', StructType([StructField('nested_int', IntegerType(), True)])),
            StructField('string', StringType(), False),
            StructField('string2', StringType(), False),
            StructField('timestamp', TimestampType(), False),
        ])

    dataframe = spark.createDataFrame(rows, schema)
    dataframe. \
        coalesce(num_files). \
        write.option('compression', 'none'). \
        mode('overwrite'). \
        partitionBy(*partition_by). \
        parquet(output_url)

    if shutdown:
        spark.stop()

    return expected_data
Creates a dataset in tmp_url location. The dataset emulates non-petastorm dataset, i.e. contains only native
    parquet types.

    These are the fields with mock data:
      'id', 'int_fixed_size_list', 'datetime', 'timestamp', 'string', 'string2', 'float64'

    :param output_url: Url specifying the location the parquet store is written to
    :param num_rows: Number of rows in the generated dataset
    :param num_files: Number of parquet files that will be written into the parquet store
    :param spark: An instance of spark session object. If `None` (default), a new spark session is created.
    :param partition_by: A list of fields to partition the parquet store by.
    :return: A list of records with a copy of the data written to the dataset.
    
petastorm/tests/test_common.py