Skip to content

reformatted PipeUtil #1183

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sonal (sonalgoyal) merged 22 commits into from
Sep 2, 2025
Merged

reformatted PipeUtil #1183

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
de5af12
reformatted PipeUtil
Nitish1814 Jul 16, 2025
841cc63
break reader and writer as two separate components
Nitish1814 Jul 21, 2025
29433fc
changes
nitish2018 Aug 3, 2025
be3043d
changed variable scope
nitish2018 Aug 5, 2025
27f40dc
added Helper class for initializing reader
nitish2018 Aug 5, 2025
82b580a
updated WriterStrategyFactory
nitish2018 Aug 10, 2025
5242337
updated PipeUtilBase
nitish2018 Aug 11, 2025
f30609c
renamed interfaces names and added Helper to for initializing writer
nitish2018 Aug 11, 2025
657d82d
removed loader with string
nitish2018 Aug 12, 2025
2666694
added default reader and writer strategy
nitish2018 Aug 13, 2025
6255a7f
removed inMemory read strategy
nitish2018 Aug 19, 2025
df44085
Merge branch 'main' into pipeutil-refactor
Nitish1814 Aug 19, 2025
aee56b3
changed location to path property
nitish2018 Aug 19, 2025
dab4327
removed in-memory pipes
nitish2018 Aug 19, 2025
67f1b65
made variable protected
nitish2018 Aug 20, 2025
f676c36
updated strategy reader
nitish2018 Aug 25, 2025
61ea072
added back compatibility for location
nitish2018 Aug 26, 2025
3fd0c1e
removed helper from reader and writer
nitish2018 Aug 26, 2025
fb90d33
added pipe to writer strategy
nitish2018 Aug 26, 2025
10abd7c
updated snowflake format
nitish2018 Aug 26, 2025
2f5ef0c
reverted back snowflake format
nitish2018 Aug 27, 2025
a38c7a2
added constant location and path
nitish2018 Aug 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 0 additions & 9 deletions common/client/src/main/java/zingg/common/client/Client.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
protected ClientOptions options;
protected S session;
protected PipeUtilBase<S,D,R,C> pipeUtil;
public static final Log LOG = LogFactory.getLog(Client.class);

Check warning on line 39 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger should be defined private static final and have the correct class 

A logger should normally be defined private static final and be associated with the correct class.
`private final Log log;` is also allowed for rare cases where loggers need to be passed around,
with the restriction that the logger needs to be passed into the constructor.

ProperLogger (Priority: 3, Ruleset: Error Prone)
https://docs.pmd-code.org/snapshot/pmd_rules_java_errorprone.html#properlogger
protected String zFactoryClassName;


Expand Down Expand Up @@ -64,7 +64,7 @@
}
catch (Exception e) {
e.printStackTrace();
throw new ZinggClientException("An error has occured while setting up the client" + e.getMessage());

Check warning on line 67 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Thrown exception does not preserve the stack trace of exception 'e' on all code paths 

Reports exceptions that are thrown from within a catch block, yet don't refer to the
exception parameter declared by that catch block. The stack trace of the original
exception could be lost, which makes the thrown exception less informative.

To preserve the stack trace, the original exception may be used as the cause of
the new exception, using `Throwable#initCause`, or passed as a constructor argument
to the new exception. It may also be preserved using `Throwable#addSuppressed`.
The rule actually assumes that any method or constructor that takes the original
exception as argument preserves the original stack trace.

The rule allows `InvocationTargetException` and `PrivilegedActionException` to be
replaced by their cause exception. The discarded part of the stack trace is in those
cases only JDK-internal code, which is not very useful. The rule also ignores exceptions
whose name starts with `ignored`.

PreserveStackTrace (Priority: 3, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#preservestacktrace
}
}

Expand All @@ -80,13 +80,13 @@
public Client(IZArgs args, ClientOptions options, S s, String zFactory) throws ZinggClientException {
this(args, options, zFactory);
this.session = s;
LOG.debug("Session passed is " + s);

Check failure on line 83 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
if (session != null) zingg.setSession(session);

Check warning on line 84 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

This statement should have braces 

Enforce a policy for braces on control statements. It is recommended to use braces on 'if ... else'
statements and loop statements, even if they are optional. This usually makes the code clearer, and
helps prepare the future when you need to add another statement. That said, this rule lets you control
which statements are required to have braces via properties.

From 6.2.0 on, this rule supersedes WhileLoopMustUseBraces, ForLoopMustUseBraces, IfStmtMustUseBraces,
and IfElseStmtMustUseBraces.

ControlStatementBraces (Priority: 3, Ruleset: Code Style)
https://docs.pmd-code.org/snapshot/pmd_rules_java_codestyle.html#controlstatementbraces
}


public IZinggFactory getZinggFactory() throws InstantiationException, IllegalAccessException, ClassNotFoundException{
LOG.debug("z factory is " + getZFactoryClassName());

Check failure on line 89 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
return (IZinggFactory) Class.forName(getZFactoryClassName()).newInstance();
}

Expand All @@ -111,7 +111,7 @@

public void buildAndSetArguments(IZArgs args, ClientOptions options) {
setOptions(options);
int jobId = new Long(System.currentTimeMillis()).intValue();

Check warning on line 114 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Do not use `new Long(...)`, prefer `Long.valueOf(...)` 

Reports usages of primitive wrapper constructors. They are deprecated
since Java 9 and should not be used. Even before Java 9, they can
be replaced with usage of the corresponding static `valueOf` factory method
(which may be automatically inserted by the compiler since Java 1.5).
This has the advantage that it may reuse common instances instead of creating
a new instance each time.

Note that for `Boolean`, the named constants `Boolean.TRUE` and `Boolean.FALSE`
are preferred instead of `Boolean.valueOf`.

PrimitiveWrapperInstantiation (Priority: 3, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#primitivewrapperinstantiation
if (options.get(options.JOBID)!= null) {
LOG.info("Using job id from command line");
String j = options.get(options.JOBID).value;
Expand All @@ -119,7 +119,7 @@
args.setJobId(jobId);
}
else if (args.getJobId() != -1) {
jobId = (args).getJobId();

Check warning on line 122 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Useless parentheses. 

Parenthesized expressions are used to override the default operator precedence
rules. Parentheses whose removal would not change the relative nesting of operators
are unnecessary, because they don't change the semantics of the enclosing expression.

Some parentheses that strictly speaking are unnecessary, may still be considered useful
for readability. This rule allows to ignore violations on two kinds of unnecessary parentheses:
- "Clarifying" parentheses, which separate operators of difference precedence. While
unnecessary, they make precedence rules explicit, which may be useful for rarely used
operators. For example:
```java
    (a + b) & c // is equivalent to `a + b & c`, but probably clearer
```
Unset the property `ignoreClarifying` to report them.

- "Balancing" parentheses, which are unnecessary but visually balance out another pair
of parentheses around an equality operator. For example, those two expressions are equivalent:
```java
    (a == null) != (b == null)
    a == null != (b == null)
```
The parentheses on the right are required, and the parentheses on the left are
just more visually pleasing. Unset the property `ignoreBalancing` to report them.


UselessParentheses (Priority: 4, Ruleset: Code Style)
https://docs.pmd-code.org/snapshot/pmd_rules_java_codestyle.html#uselessparentheses
}

//override value of zinggDir passed from command line
Expand Down Expand Up @@ -154,12 +154,12 @@
LOG.info("");
LOG.info("**************************************************************************");
LOG.info("* *");
LOG.info("* "+getProductName()+" *");

Check failure on line 157 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
LOG.info("* (C) 2021 Zingg Labs, Inc. *");
LOG.info("* *");
LOG.info("* https://www.zingg.ai/ *");
LOG.info("* *");
LOG.info("* using: Zingg v"+getProductVersion()+" *");

Check failure on line 162 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
LOG.info("* *");
if(collectMetrics) {
LOG.info("* ** Note about analytics collection by Zingg AI ** *");
Expand Down Expand Up @@ -195,12 +195,12 @@
boolean success = true;
try {

for (String a: args) LOG.debug("args " + a);

Check failure on line 198 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement

Check warning on line 198 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

This statement should have braces 

Enforce a policy for braces on control statements. It is recommended to use braces on 'if ... else'
statements and loop statements, even if they are optional. This usually makes the code clearer, and
helps prepare the future when you need to add another statement. That said, this rule lets you control
which statements are required to have braces via properties.

From 6.2.0 on, this rule supersedes WhileLoopMustUseBraces, ForLoopMustUseBraces, IfStmtMustUseBraces,
and IfElseStmtMustUseBraces.

ControlStatementBraces (Priority: 3, Ruleset: Code Style)
https://docs.pmd-code.org/snapshot/pmd_rules_java_codestyle.html#controlstatementbraces
options = getClientOptions(args);
setOptions(options);

if (options.has(options.HELP) || options.has(options.HELP1) || options.get(ClientOptions.PHASE) == null) {
LOG.warn(options.getHelp());

Check failure on line 203 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
System.exit(0);
}
String phase = options.get(ClientOptions.PHASE).value.trim();
Expand All @@ -214,7 +214,7 @@

LOG.warn("Zingg processing has completed");
}
catch(Throwable throwable) {

Check warning on line 217 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

A catch statement should never catch throwable since it includes errors. 

Catching Throwable errors is not recommended since its scope is very broad. It includes runtime issues such as
OutOfMemoryError that should be exposed and managed separately.

AvoidCatchingThrowable (Priority: 3, Ruleset: Error Prone)
https://docs.pmd-code.org/snapshot/pmd_rules_java_errorprone.html#avoidcatchingthrowable
success = false;

if (options != null && options.get(ClientOptions.EMAIL) != null) {
Expand All @@ -222,9 +222,9 @@
"Zingg Error ",
throwable.getMessage()));
}
LOG.warn("Apologies for this message. Zingg has encountered an error. "
+ throwable.getMessage());

Check failure on line 226 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

Logger calls should be surrounded by log level guards. 

Whenever using a log level, one should check if it is actually enabled, or
otherwise skip the associate String creation and manipulation, as well as any method calls.

An alternative to checking the log level are substituting parameters, formatters or lazy logging
with lambdas. The available alternatives depend on the actual logging framework.

GuardLogStatement (Priority: 2, Ruleset: Best Practices)
https://docs.pmd-code.org/snapshot/pmd_rules_java_bestpractices.html#guardlogstatement
if (LOG.isDebugEnabled()) throwable.printStackTrace();

Check warning on line 227 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

This statement should have braces 

Enforce a policy for braces on control statements. It is recommended to use braces on 'if ... else'
statements and loop statements, even if they are optional. This usually makes the code clearer, and
helps prepare the future when you need to add another statement. That said, this rule lets you control
which statements are required to have braces via properties.

From 6.2.0 on, this rule supersedes WhileLoopMustUseBraces, ForLoopMustUseBraces, IfStmtMustUseBraces,
and IfElseStmtMustUseBraces.

ControlStatementBraces (Priority: 3, Ruleset: Code Style)
https://docs.pmd-code.org/snapshot/pmd_rules_java_codestyle.html#controlstatementbraces
}
finally {
cleanupAndExit(success, client);
Expand Down Expand Up @@ -259,7 +259,7 @@
printBanner(arguments.getCollectMetrics());
zingg.setClientOptions(getOptions());
zingg.init(getArguments(), getSession(),getOptions());
if (session != null) zingg.setSession(session);

Check warning on line 262 in common/client/src/main/java/zingg/common/client/Client.java

View workflow job for this annotation

GitHub Actions / PMD Static Code Analysis

This statement should have braces 

Enforce a policy for braces on control statements. It is recommended to use braces on 'if ... else'
statements and loop statements, even if they are optional. This usually makes the code clearer, and
helps prepare the future when you need to add another statement. That said, this rule lets you control
which statements are required to have braces via properties.

From 6.2.0 on, this rule supersedes WhileLoopMustUseBraces, ForLoopMustUseBraces, IfStmtMustUseBraces,
and IfElseStmtMustUseBraces.

ControlStatementBraces (Priority: 3, Ruleset: Code Style)
https://docs.pmd-code.org/snapshot/pmd_rules_java_codestyle.html#controlstatementbraces
initializeListeners();
EventsListener.getInstance().fireEvent(new ZinggStartEvent());
}
Expand Down Expand Up @@ -297,15 +297,6 @@
this.options = options;
}



protected ArgumentsUtil<?> getArgsUtil(String phase) {
if (argsUtil==null) {
argsUtil = new ArgumentsUtil(Arguments.class);
}
return argsUtil;
}

protected void cleanupAndExit(boolean success, Client<S,D,R,C,T> client) {
if (!success) {
EventsListener.getInstance().fireEvent(new ZinggFailEvent());
Expand Down
3 changes: 1 addition & 2 deletions common/client/src/main/java/zingg/common/client/pipe/FilePipe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
package zingg.common.client.pipe;

public class FilePipe {

public static final String LOCATION = "location";
public static final String PATH = "path";
public static final String HEADER = "header";
public static final String DELIMITER = "delimiter";
public static final String TABLE = "table";

}
21 changes: 0 additions & 21 deletions common/client/src/main/java/zingg/common/client/util/DFReader.java
View file
Open in desktop

This file was deleted.

11 changes: 0 additions & 11 deletions common/client/src/main/java/zingg/common/client/util/DFWriter.java
View file
Open in desktop

This file was deleted.

276 changes: 22 additions & 254 deletions common/client/src/main/java/zingg/common/client/util/PipeUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,295 +8,63 @@

import zingg.common.client.ZFrame;
import zingg.common.client.ZinggClientException;
import zingg.common.client.pipe.FilePipe;
import zingg.common.client.pipe.Pipe;
import zingg.common.client.util.reader.PipeUtilReader;
import zingg.common.client.util.writer.PipeUtilWriter;

//import com.datastax.spark.connector.cql.*;
//import org.elasticsearch.spark.sql.api.java.JavaEsSparkSQL;
//import zingg.scala.DFUtil;

public abstract class PipeUtil<S,D,R,C> implements PipeUtilBase<S,D,R,C>{
public abstract class PipeUtil<S,D,R,C> implements PipeUtilBase<S,D,R,C> {

protected S session;
public final Log LOG = LogFactory.getLog(PipeUtil.class);
protected final PipeUtilWriter<D, R, C> pipeUtilWriter;
protected final PipeUtilReader<S, D, R, C> pipeUtilReader;

public final Log LOG = LogFactory.getLog(PipeUtil.class);

public PipeUtil(S spark) {
public PipeUtil(S spark, PipeUtilWriter<D, R, C> pipeUtilWriter, PipeUtilReader<S, D, R, C> pipeUtilReader) {
this.session = spark;
this.pipeUtilWriter = pipeUtilWriter;
this.pipeUtilReader = pipeUtilReader;
}


@Override
public S getSession(){
return this.session;
}

@Override
public void setSession(S session){
this.session = session;
}

//public abstract DFReader<S,D,R,C> getReader();//getSession().read()

public DFReader<D,R,C> getReader(Pipe<D,R,C> p) {
DFReader<D,R,C> reader = getReader();
reader = reader.format(p.getFormat());
if (p.getSchema() != null) {
reader = reader.setSchema(p.getSchema()); //.schema(StructType.fromDDL(p.getSchema()));
}

for (String key : p.getProps().keySet()) {
reader = reader.option(key, p.get(key));
}
reader = reader.option("mode", "PERMISSIVE");
return reader;
}

public ZFrame<D,R,C> getInput(Pipe<D,R,C> p, DFReader<D,R,C> reader) throws ZinggClientException{
ZFrame<D,R,C> input = null;
if (p.getProps().containsKey(FilePipe.LOCATION)) {
input = reader.load(p.get(FilePipe.LOCATION));
}
else {
input = reader.load();
}
return input;
}

protected ZFrame<D,R,C> read(DFReader<D,R,C> reader, Pipe<D,R,C> p, boolean addSource) throws ZinggClientException{
ZFrame<D,R,C> input = null;
LOG.warn("Reading " + p);
try {
input = getInput(p, reader);

if (addSource) {
input = input.withColumn(ColName.SOURCE_COL, p.getName());
}

p.setDataset(input);
} catch (Exception ex) {
LOG.warn(ex.getMessage());
throw new ZinggClientException("Could not read data.", ex);
}
return getZFrame(input); //new SparkFrame(input);
}

public ZFrame<D,R,C> readInternal(Pipe<D,R,C> p,
boolean addSource) throws ZinggClientException {
DFReader<D,R,C> reader = getReader(p);
return read(reader, p, addSource);
}

/*
public ZFrame<D,R,C> joinTrainingSetstoGetLabels(ZFrame<D,R,C> jdbc,
ZFrame<D,R,C> file) {
file = file.drop(ColName.MATCH_FLAG_COL);
file.printSchema();
file.show();
jdbc = jdbc.select(jdbc.col(ColName.ID_COL), jdbc.col(ColName.SOURCE_COL),jdbc.col(ColName.MATCH_FLAG_COL),
jdbc.col(ColName.CLUSTER_COLUMN));
String[] cols = jdbc.columns();
for (int i=0; i < cols.length; ++i) {
cols[i] = ColName.COL_PREFIX + cols[i];
}
jdbc = jdbc.toDF(cols).cache();

jdbc = jdbc.withColumnRenamed(ColName.COL_PREFIX + ColName.MATCH_FLAG_COL, ColName.MATCH_FLAG_COL);
jdbc.printSchema();

jdbc.show();
LOG.warn("Building labels ");
Dataset<Row> pairs = file.join(jdbc, file.col(ColName.ID_COL).equalTo(
jdbc.col(ColName.COL_PREFIX + ColName.ID_COL))
.and(file.col(ColName.SOURCE_COL).equalTo(
jdbc.col(ColName.COL_PREFIX + ColName.SOURCE_COL)))
.and(file.col(ColName.CLUSTER_COLUMN).equalTo(
jdbc.col(ColName.COL_PREFIX + ColName.CLUSTER_COLUMN))));
LOG.warn("Pairs are " + pairs.count());
//in training, we only need that record matches only with lines bigger than itself
//in the case of normal as well as in the case of linking
pairs = pairs.drop(ColName.COL_PREFIX + ColName.SOURCE_COL);
pairs = pairs.drop(ColName.COL_PREFIX + ColName.ID_COL);
pairs = pairs.drop(ColName.COL_PREFIX + ColName.CLUSTER_COLUMN);

return pairs;
}
*/

public ZFrame<D,R,C> readInternal(boolean addLineNo,
boolean addSource, Pipe<D,R,C>... pipes) throws ZinggClientException {
return readInternal(false, addLineNo,addSource, pipes);
}

public ZFrame<D,R,C> readInternal(boolean addExtraCol, boolean addLineNo,
boolean addSource, Pipe<D,R,C>... pipes) throws ZinggClientException {
ZFrame<D,R,C> input = null;

for (Pipe p : pipes) {
if (input == null) {
input = readInternal(p, addSource);
if (LOG.isDebugEnabled()) {
LOG.debug("input size is " + input.count());
}
} else {
if(!addExtraCol) {
input = input.union(readInternal(p, addSource));
} else {
input = input.unionByName(readInternal(p, addSource),true);
}
}
}
// we will probably need to create row number as string with pipename/id as
// suffix
if (addLineNo)
input = addLineNo(input); //new SparkFrame(new SparkDSUtil(getSession()).addRowNumber(input).df());
// we need to transform the input here by using stop words
return input;
}

@Override
public ZFrame<D,R,C> read(boolean addLineNo, boolean addSource, Pipe<D,R,C>... pipes) throws ZinggClientException {
ZFrame<D,R,C> rows = readInternal(addLineNo, addSource, pipes);
rows = rows.cache();
return rows;
return pipeUtilReader.read(addLineNo, addSource, pipes);
}

/*
public ZFrame<D,R,C> sample(S spark, Pipe p) throws ZinggClientException {
DataFrameReader reader = getReader(spark, p);
reader.option("inferSchema", true);
reader.option("mode", "DROPMALFORMED");
LOG.info("reader is ready to sample with inferring " + p.get(FilePipe.LOCATION));
LOG.warn("Reading input of type " + p.getFormat());
ZFrame<D,R,C> input = read(reader, p, false);
// LOG.warn("inferred schema " + input.schema());
List<Row> values = input.takeAsList(10);
values.forEach(r -> LOG.warn(r));
Dataset<Row> ret = spark.createDataFrame(values, input.schema());
return ret;
}
*/

@Override
public ZFrame<D,R,C> read(boolean addLineNo, int numPartitions,
boolean addSource, Pipe<D,R,C>... pipes) throws ZinggClientException {
return read(false, addLineNo, numPartitions, addSource, pipes);
return pipeUtilReader.read(false, addLineNo, numPartitions, addSource, pipes);
}

@Override
public ZFrame<D,R,C> read(boolean addExtraCol, boolean addLineNo, int numPartitions,
boolean addSource, Pipe<D,R,C>... pipes) throws ZinggClientException {
ZFrame<D,R,C> rows = readInternal(addExtraCol, addLineNo, addSource, pipes);
rows = rows.repartition(numPartitions);
rows = rows.cache();
return rows;
}

public void write(ZFrame<D,R,C> toWriteOrig,
Pipe<D,R,C>... pipes) throws ZinggClientException {
try {
for (Pipe<D,R,C> p: pipes) {
//Dataset<Row> toWrite = toWriteOrig.df();
//DataFrameWriter writer = toWrite.write();
DFWriter<D,R,C> writer = getWriter(toWriteOrig);

LOG.warn("Writing output " + p);

//SparkPipe sPipe = (SparkPipe) p;
if (p.getMode() != null) {
writer.setMode(p.getMode()); //SaveMode.valueOf(p.getMode()));
}
else {
writer.setMode("Append"); //SaveMode.valueOf("Append"));
}
writer = getWriterWithFormat(writer, p);

for (String key: p.getProps().keySet()) {
writer = writer.option(key, p.get(key));
}
save(p, writer, toWriteOrig);
}
} catch (Exception ex) {
throw new ZinggClientException(ex.getMessage());
}
return pipeUtilReader.read(addExtraCol, addLineNo, numPartitions, addSource, pipes);
}

public DFWriter<D,R,C> getWriterWithFormat(DFWriter<D,R,C> writer, Pipe<D,R,C> p) {
writer = writer.format(p.getFormat());
return writer;
@Override
public void write(ZFrame<D, R, C> toWriteOrig, Pipe<D, R, C>... pipes) throws ZinggClientException {
pipeUtilWriter.write(toWriteOrig, pipes);
}

public void save(Pipe<D,R,C> p, DFWriter<D,R,C> writer, ZFrame<D,R,C> toWriteOrig){
if (p.getProps().containsKey("location")) {
LOG.warn("Writing file");
writer.save(p.get(FilePipe.LOCATION));
}
else{
writer.save();
}
}
/*
public void writePerSource(Dataset<Row> toWrite, Arguments args, JavaSparkContext ctx, Pipe[] pipes ) throws ZinggClientException {
List<Row> sources = toWrite.select(ColName.SOURCE_COL).distinct().collectAsList();
for (Row r : sources) {
Dataset<Row> toWriteNow = toWrite.filter(toWrite.col(ColName.SOURCE_COL).equalTo(r.get(0)));
toWriteNow = toWriteNow.drop(ColName.SOURCE_COL);
write(toWriteNow, args, ctx, pipes);

}
}
*/



@Override
public String getPipesAsString(Pipe<D,R,C>[] pipes) {
return Arrays.stream(pipes)
.map(p -> p.getFormat())
.map(Pipe::getFormat)
.collect(Collectors.toList())
.stream().reduce((p1, p2) -> p1 + "," + p2)
.map(Object::toString)
.orElse("");
}




/*
* public String getTableCreateCQL(Pipe p, Dataset<Row> df) {
*
* Set<String> partitionKeys = new TreeSet<String>() { {
* add(p.get(CassandraPipe.PRIMARY_KEY)); } }; int c = 0; Map<String, Integer>
* clustereingKeys = new TreeMap<String, Integer>(); if (p.getAddProps()!= null
* && p.getAddProps().containsKey("clusterBy")) { for (String clBy :
* p.getAddProps().get("clusterBy").split(",")) { { clustereingKeys.put(clBy,
* c++); } } }
*
* TableDef td = TableDef.fromDataFrame(df, p.get(CassandraPipe.KEYSPACE),
* p.get(CassandraPipe.TABLE), ProtocolVersion.NEWEST_SUPPORTED);
*
* List<ColumnDef> partKeyList = new ArrayList<ColumnDef>(); List<ColumnDef>
* clusterColumnList = new ArrayList<ColumnDef>(); List<ColumnDef>
* regColulmnList = new ArrayList<ColumnDef>();
*
* scala.collection.Iterator<ColumnDef> iter = td.allColumns() .iterator();
* while (iter.hasNext()) { ColumnDef col = iter.next(); String colName =
* col.columnName(); if (partitionKeys.contains(colName)) { partKeyList.add(new
* ColumnDef(colName, PartitionKeyColumn$.MODULE$, col.columnType())); } else if
* (clustereingKeys.containsKey(colName)) { int idx =
* clustereingKeys.get(colName); clusterColumnList.add(new ColumnDef(colName,
* new ClusteringColumn(idx), col.columnType())); } else { if
* (colName.equals("dob")) { LOG.warn("yes dob"); regColulmnList.add(new
* ColumnDef(colName, new StaticColumn$.MODULE$, col.columnType())); } else {
* regColulmnList.add(new ColumnDef(colName, RegularColumn$.MODULE$,
* col.columnType())); } } }
*
* TableDef newTd = new TableDef(td.keyspaceName(), td.tableName(),
* JavaConverters.asScalaIteratorConverter(partKeyList.iterator()).asScala().
* toSeq(),
* JavaConverters.asScalaIteratorConverter(clusterColumnList.iterator()).asScala
* ().toSeq() ,
* JavaConverters.asScalaIteratorConverter(regColulmnList.iterator()).asScala().
* toSeq(), td.indexes(), td.isView()); String cql = newTd.cql();
* System.out.println(cql); System.out.println(cql.replace(",dob", newChar));
*
* return cql;
*
* }
*/

}
Loading
Loading