CsvSource implements ISource { @SuppressWarnings("unchecked") @Override public <I> I adaptToBuildEngine(Class<I> engineInterface) { return (I) new NSparkCubingSource() { @Override public Dataset<Row> getSourceData(TableDesc table, SparkSession ss, Map<String, String> parameters) { String path = null; KylinConfig kylinConfig = KylinBuildEnv.get().kylinConfig(); String separator; boolean withHeader = false; if (kylinConfig.getDeployEnv().equals("UT") && (parameters != null && parameters.get("separator") == null)) { path = "file: "../../examples/test_case_data/parquet_test/data/" + table.identity() + ".csv") .getAbsolutePath(); separator = ""; } else { String project = parameters.get("project") == null ? "" : parameters.get("project") + "/"; path = KylinBuildEnv.get().kylinConfig().getHdfsWorkingDirectory() + project + "csv/" + table.identity() + ".csv"; separator = parameters.get("separator"); if (parameters.get("withHeader") != null) { withHeader = Boolean.parseBoolean(parameters.get("withHeader")); } } switch (separator) { case "space": separator = " "; break; case "tab": separator = "\t"; break; default: separator = ","; } Dataset<Row> delimiter = ss.read().option("delimiter", separator) .option("header", withHeader).schema(table.toSchema()).csv(path); return delimiter; } }; } CsvSource();  CsvSource(KylinConfig config); @SuppressWarnings("unchecked") @Override I adaptToBuildEngine(Class<I> engineInterface); @Override SourcePartition enrichSourcePartitionBeforeBuild(IBuildable buildable, SourcePartition srcPartition); @Override ISourceMetadataExplorer getSourceMetadataExplorer(); @Override IReadableTable createReadableTable(org.apache.kylin.metadata.model.TableDesc tableDesc, String uuid); @Override ISampleDataDeployer getSampleDataDeployer(); @Override void unloadTable(String tableName, String project); @Override void close();  }
@Test public void testGetSourceDataFromFactTable() { CubeManager cubeMgr = CubeManager.getInstance(getTestConfig()); CubeInstance cube = cubeMgr.getCube(CUBE_NAME); TableDesc fact = MetadataConverter.extractFactTable(cube); List<ColumnDesc> colDescs = Lists.newArrayList(); Iterator<ColumnDesc> iterator = fact.columns().iterator(); while (iterator.hasNext()) { colDescs.add(iterator.next()); } NSparkCubingEngine.NSparkCubingSource cubingSource = new CsvSource().adaptToBuildEngine(NSparkCubingEngine.NSparkCubingSource.class); Dataset<Row> cubeDS = cubingSource.getSourceData(fact, ss, Maps.newHashMap()); cubeDS.take(10); StructType schema = cubeDS.schema(); for (int i = 0; i < colDescs.size(); i++) { StructField field = schema.fields()[i]; Assert.assertEquals(field.name(), colDescs.get(i).columnName()); Assert.assertEquals(field.dataType(), colDescs.get(i).dataType()); } }
@Test public void testGetSourceDataFromLookupTable() { CubeManager cubeMgr = CubeManager.getInstance(getTestConfig()); CubeInstance cube = cubeMgr.getCube(CUBE_NAME); Iterator<TableDesc> iterator = MetadataConverter.extractLookupTable(cube).iterator(); while (iterator.hasNext()) { TableDesc lookup = iterator.next(); NSparkCubingEngine.NSparkCubingSource cubingSource = new CsvSource().adaptToBuildEngine(NSparkCubingEngine.NSparkCubingSource.class); Dataset<Row> sourceData = cubingSource.getSourceData(lookup, ss, Maps.newHashMap()); List<Row> rows = sourceData.collectAsList(); Assert.assertTrue(rows != null && rows.size() > 0); } }